Skip to content

Commit

Permalink
Bugfix/fix interceptor/webrtc unit test memory leak (#629)
Browse files Browse the repository at this point in the history
* Fix MockStream cyclic dependency to itself

- it often binds itself into Interceptor and owns the returned sender/writer. The returned sender/writer might contain the Arc<> to MockStream itself. Thus, the cyclic dependency is formed. We create an internal struct to avoid this.

* Fix cyclic dependency between PeerConnectionInternal and StatsInterceptor

- PeerConnectionInternal should not own StatsInterceptor. Make it Weak<>

* Fix cyclic dependency for dtls_transport

- IceTransport holds OnConnectionStateChangeFn, which hods DtlsTransport and DtlsTransport holds IceTransport. In the callback we should use Weak<>

* Fix self reference test case in data_channel_test.rs

* Fix formatting
mutexd authored Nov 10, 2024

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature.
1 parent 03237d0 commit 8db3679
Showing 4 changed files with 250 additions and 230 deletions.
72 changes: 38 additions & 34 deletions interceptor/src/mock/mock_stream.rs
Original file line number Diff line number Diff line change
@@ -17,6 +17,10 @@ pub struct MockStream {
rtcp_writer: Mutex<Option<Arc<dyn RTCPWriter + Send + Sync>>>,
rtp_writer: Mutex<Option<Arc<dyn RTPWriter + Send + Sync>>>,

internal: Arc<MockStreamInternal>,
}

struct MockStreamInternal {
rtcp_out_modified_tx: mpsc::Sender<RTCPPackets>,
rtp_out_modified_tx: mpsc::Sender<rtp::packet::Packet>,
rtcp_in_rx: Mutex<mpsc::Receiver<RTCPPackets>>,
@@ -46,44 +50,44 @@ impl MockStream {

let stream = Arc::new(MockStream {
interceptor: Arc::clone(&interceptor),

rtcp_writer: Mutex::new(None),
rtp_writer: Mutex::new(None),

rtcp_in_tx: Mutex::new(Some(rtcp_in_tx)),
rtp_in_tx: Mutex::new(Some(rtp_in_tx)),
rtcp_in_rx: Mutex::new(rtcp_in_rx),
rtp_in_rx: Mutex::new(rtp_in_rx),

rtcp_out_modified_tx,
rtp_out_modified_tx,
rtcp_out_modified_rx: Mutex::new(rtcp_out_modified_rx),
rtp_out_modified_rx: Mutex::new(rtp_out_modified_rx),

rtcp_in_modified_rx: Mutex::new(rtcp_in_modified_rx),
rtp_in_modified_rx: Mutex::new(rtp_in_modified_rx),
internal: Arc::new(MockStreamInternal {
rtcp_in_tx: Mutex::new(Some(rtcp_in_tx)),
rtp_in_tx: Mutex::new(Some(rtp_in_tx)),
rtcp_in_rx: Mutex::new(rtcp_in_rx),
rtp_in_rx: Mutex::new(rtp_in_rx),

rtcp_out_modified_tx,
rtp_out_modified_tx,
rtcp_out_modified_rx: Mutex::new(rtcp_out_modified_rx),
rtp_out_modified_rx: Mutex::new(rtp_out_modified_rx),

rtcp_in_modified_rx: Mutex::new(rtcp_in_modified_rx),
rtp_in_modified_rx: Mutex::new(rtp_in_modified_rx),
}),
});

let rtcp_writer = interceptor
.bind_rtcp_writer(Arc::clone(&stream) as Arc<dyn RTCPWriter + Send + Sync>)
.bind_rtcp_writer(Arc::clone(&stream.internal) as Arc<dyn RTCPWriter + Send + Sync>)
.await;
{
let mut rw = stream.rtcp_writer.lock().await;
*rw = Some(rtcp_writer);
*rw = Some(Arc::clone(&rtcp_writer));
}
let rtp_writer = interceptor
.bind_local_stream(
info,
Arc::clone(&stream) as Arc<dyn RTPWriter + Send + Sync>,
Arc::clone(&stream.internal) as Arc<dyn RTPWriter + Send + Sync>,
)
.await;
{
let mut rw = stream.rtp_writer.lock().await;
*rw = Some(rtp_writer);
*rw = Some(Arc::clone(&rtp_writer));
}

let rtcp_reader = interceptor
.bind_rtcp_reader(Arc::clone(&stream) as Arc<dyn RTCPReader + Send + Sync>)
.bind_rtcp_reader(Arc::clone(&stream.internal) as Arc<dyn RTCPReader + Send + Sync>)
.await;
tokio::spawn(async move {
let mut buf = vec![0u8; 1500];
@@ -104,7 +108,7 @@ impl MockStream {
let rtp_reader = interceptor
.bind_remote_stream(
info,
Arc::clone(&stream) as Arc<dyn RTPReader + Send + Sync>,
Arc::clone(&stream.internal) as Arc<dyn RTPReader + Send + Sync>,
)
.await;
tokio::spawn(async move {
@@ -153,23 +157,23 @@ impl MockStream {

/// receive_rtcp schedules a new rtcp batch, so it can be read be the stream
pub async fn receive_rtcp(&self, pkts: Vec<Box<dyn rtcp::packet::Packet + Send + Sync>>) {
let rtcp_in_tx = self.rtcp_in_tx.lock().await;
let rtcp_in_tx = self.internal.rtcp_in_tx.lock().await;
if let Some(tx) = &*rtcp_in_tx {
let _ = tx.send(pkts).await;
}
}

/// receive_rtp schedules a rtp packet, so it can be read be the stream
pub async fn receive_rtp(&self, pkt: rtp::packet::Packet) {
let rtp_in_tx = self.rtp_in_tx.lock().await;
let rtp_in_tx = self.internal.rtp_in_tx.lock().await;
if let Some(tx) = &*rtp_in_tx {
let _ = tx.send(pkt).await;
}
}

/// written_rtcp returns a channel containing the rtcp batches written, modified by the interceptor
pub async fn written_rtcp(&self) -> Option<Vec<Box<dyn rtcp::packet::Packet + Send + Sync>>> {
let mut rtcp_out_modified_rx = self.rtcp_out_modified_rx.lock().await;
let mut rtcp_out_modified_rx = self.internal.rtcp_out_modified_rx.lock().await;
rtcp_out_modified_rx.recv().await
}

@@ -180,7 +184,7 @@ impl MockStream {
&self,
) -> Option<Vec<Box<dyn rtcp::packet::Packet + Send + Sync>>> {
let mut last = None;
let mut rtcp_out_modified_rx = self.rtcp_out_modified_rx.lock().await;
let mut rtcp_out_modified_rx = self.internal.rtcp_out_modified_rx.lock().await;

while let Ok(v) = rtcp_out_modified_rx.try_recv() {
last = Some(v);
@@ -191,40 +195,40 @@ impl MockStream {

/// written_rtp returns a channel containing rtp packets written, modified by the interceptor
pub async fn written_rtp(&self) -> Option<rtp::packet::Packet> {
let mut rtp_out_modified_rx = self.rtp_out_modified_rx.lock().await;
let mut rtp_out_modified_rx = self.internal.rtp_out_modified_rx.lock().await;
rtp_out_modified_rx.recv().await
}

/// read_rtcp returns a channel containing the rtcp batched read, modified by the interceptor
pub async fn read_rtcp(
&self,
) -> Option<Result<Vec<Box<dyn rtcp::packet::Packet + Send + Sync>>>> {
let mut rtcp_in_modified_rx = self.rtcp_in_modified_rx.lock().await;
let mut rtcp_in_modified_rx = self.internal.rtcp_in_modified_rx.lock().await;
rtcp_in_modified_rx.recv().await
}

/// read_rtp returns a channel containing the rtp packets read, modified by the interceptor
pub async fn read_rtp(&self) -> Option<Result<rtp::packet::Packet>> {
let mut rtp_in_modified_rx = self.rtp_in_modified_rx.lock().await;
let mut rtp_in_modified_rx = self.internal.rtp_in_modified_rx.lock().await;
rtp_in_modified_rx.recv().await
}

/// close closes the stream and the underlying interceptor
/// close closes the stream
pub async fn close(&self) -> Result<()> {
{
let mut rtcp_in_tx = self.rtcp_in_tx.lock().await;
let mut rtcp_in_tx = self.internal.rtcp_in_tx.lock().await;
rtcp_in_tx.take();
}
{
let mut rtp_in_tx = self.rtp_in_tx.lock().await;
let mut rtp_in_tx = self.internal.rtp_in_tx.lock().await;
rtp_in_tx.take();
}
self.interceptor.close().await
}
}

#[async_trait]
impl RTCPWriter for MockStream {
impl RTCPWriter for MockStreamInternal {
async fn write(
&self,
pkts: &[Box<dyn rtcp::packet::Packet + Send + Sync>],
@@ -237,7 +241,7 @@ impl RTCPWriter for MockStream {
}

#[async_trait]
impl RTCPReader for MockStream {
impl RTCPReader for MockStreamInternal {
async fn read(
&self,
buf: &mut [u8],
@@ -260,15 +264,15 @@ impl RTCPReader for MockStream {
}

#[async_trait]
impl RTPWriter for MockStream {
impl RTPWriter for MockStreamInternal {
async fn write(&self, pkt: &rtp::packet::Packet, _a: &Attributes) -> Result<usize> {
let _ = self.rtp_out_modified_tx.send(pkt.clone()).await;
Ok(0)
}
}

#[async_trait]
impl RTPReader for MockStream {
impl RTPReader for MockStreamInternal {
async fn read(
&self,
buf: &mut [u8],
26 changes: 17 additions & 9 deletions webrtc/src/data_channel/data_channel_test.rs
Original file line number Diff line number Diff line change
@@ -202,11 +202,15 @@ async fn test_data_channel_send_before_signaling() -> Result<()> {
return Box::pin(async {});
}
Box::pin(async move {
let d2 = Arc::clone(&d);
let d2 = Arc::downgrade(&d);
d.on_message(Box::new(move |_: DataChannelMessage| {
let d3 = Arc::clone(&d2);
let d3 = d2.clone();
Box::pin(async move {
let result = d3.send(&Bytes::from(b"Pong".to_vec())).await;
let result = d3
.upgrade()
.unwrap()
.send(&Bytes::from(b"Pong".to_vec()))
.await;
assert!(result.is_ok(), "Failed to send string on data channel");
})
}));
@@ -218,11 +222,11 @@ async fn test_data_channel_send_before_signaling() -> Result<()> {

assert!(dc.ordered(), "Ordered should be set to true");

let dc2 = Arc::clone(&dc);
let dc2 = Arc::downgrade(&dc);
dc.on_open(Box::new(move || {
let dc3 = Arc::clone(&dc2);
let dc3 = dc2.clone();
Box::pin(async move {
let result = dc3.send_text("Ping".to_owned()).await;
let result = dc3.upgrade().unwrap().send_text("Ping".to_owned()).await;
assert!(result.is_ok(), "Failed to send string on data channel");
})
}));
@@ -258,12 +262,16 @@ async fn test_data_channel_send_after_connected() -> Result<()> {
return Box::pin(async {});
}
Box::pin(async move {
let d2 = Arc::clone(&d);
let d2 = Arc::downgrade(&d);
d.on_message(Box::new(move |_: DataChannelMessage| {
let d3 = Arc::clone(&d2);
let d3 = d2.clone();

Box::pin(async move {
let result = d3.send(&Bytes::from(b"Pong".to_vec())).await;
let result = d3
.upgrade()
.unwrap()
.send(&Bytes::from(b"Pong".to_vec()))
.await;
assert!(result.is_ok(), "Failed to send string on data channel");
})
}));
10 changes: 7 additions & 3 deletions webrtc/src/peer_connection/mod.rs
Original file line number Diff line number Diff line change
@@ -236,9 +236,13 @@ impl RTCPeerConnection {
};

let weak_interceptor = Arc::downgrade(&interceptor);
let (internal, configuration) =
PeerConnectionInternal::new(api, weak_interceptor, stats_interceptor, configuration)
.await?;
let (internal, configuration) = PeerConnectionInternal::new(
api,
weak_interceptor,
Arc::downgrade(&stats_interceptor),
configuration,
)
.await?;
let internal_rtcp_writer = Arc::clone(&internal) as Arc<dyn RTCPWriter + Send + Sync>;
let interceptor_rtcp_writer = interceptor.bind_rtcp_writer(internal_rtcp_writer).await;

372 changes: 188 additions & 184 deletions webrtc/src/peer_connection/peer_connection_internal.rs
Original file line number Diff line number Diff line change
@@ -1,12 +1,6 @@
use std::collections::VecDeque;
use std::sync::Weak;

use arc_swap::ArcSwapOption;
use portable_atomic::AtomicIsize;
use smol_str::SmolStr;
use tokio::time::Instant;
use util::Unmarshal;

use super::*;
use crate::rtp_transceiver::create_stream_info;
use crate::stats::stats_collector::StatsCollector;
@@ -17,6 +11,11 @@ use crate::stats::{
use crate::track::track_local::track_local_static_sample::TrackLocalStaticSample;
use crate::track::TrackStream;
use crate::SDP_ATTRIBUTE_RID;
use arc_swap::ArcSwapOption;
use portable_atomic::AtomicIsize;
use smol_str::SmolStr;
use tokio::time::Instant;
use util::Unmarshal;

pub(crate) struct PeerConnectionInternal {
/// a value containing the last known greater mid value
@@ -67,14 +66,14 @@ pub(crate) struct PeerConnectionInternal {
pub(super) setting_engine: Arc<SettingEngine>,
pub(crate) media_engine: Arc<MediaEngine>,
pub(super) interceptor: Weak<dyn Interceptor + Send + Sync>,
stats_interceptor: Arc<stats::StatsInterceptor>,
stats_interceptor: Weak<stats::StatsInterceptor>,
}

impl PeerConnectionInternal {
pub(super) async fn new(
api: &API,
interceptor: Weak<dyn Interceptor + Send + Sync>,
stats_interceptor: Arc<stats::StatsInterceptor>,
stats_interceptor: Weak<stats::StatsInterceptor>,
mut configuration: RTCConfiguration,
) -> Result<(Arc<Self>, RTCConfiguration)> {
// Create the ice gatherer
@@ -137,7 +136,7 @@ impl PeerConnectionInternal {
let ice_connection_state = Arc::clone(&pc.ice_connection_state);
let peer_connection_state = Arc::clone(&pc.peer_connection_state);
let is_closed = Arc::clone(&pc.is_closed);
let dtls_transport = Arc::clone(&pc.dtls_transport);
let dtls_transport = Arc::downgrade(&pc.dtls_transport);
let on_ice_connection_state_change_handler =
Arc::clone(&pc.on_ice_connection_state_change_handler);
let on_peer_connection_state_change_handler =
@@ -159,7 +158,7 @@ impl PeerConnectionInternal {
}
};

let dtls_transport = Arc::clone(&dtls_transport);
let dtls_transport = dtls_transport.clone();
let ice_connection_state = Arc::clone(&ice_connection_state);
let on_ice_connection_state_change_handler =
Arc::clone(&on_ice_connection_state_change_handler);
@@ -175,15 +174,18 @@ impl PeerConnectionInternal {
)
.await;

let dtls_transport_state = dtls_transport.state();
RTCPeerConnection::update_connection_state(
&on_peer_connection_state_change_handler,
&is_closed,
&peer_connection_state,
cs,
dtls_transport_state,
)
.await;
if let Some(dtls_transport) = dtls_transport.upgrade() {
RTCPeerConnection::update_connection_state(
&on_peer_connection_state_change_handler,
&is_closed,
&peer_connection_state,
cs,
dtls_transport.state(),
)
.await;
} else {
log::warn!("on_ice_connection_state_change: dtls_transport unavailable");
}
})
},
));
@@ -1277,90 +1279,91 @@ impl PeerConnectionInternal {
}
}

let stream_stats = self
.stats_interceptor
.fetch_inbound_stats(track_infos.iter().map(|t| t.ssrc).collect())
.await;
if let Some(stats_interceptor) = self.stats_interceptor.upgrade() {
let stream_stats = stats_interceptor
.fetch_inbound_stats(track_infos.iter().map(|t| t.ssrc).collect())
.await;

for (stats, info) in
(stream_stats.into_iter().zip(track_infos)).filter_map(|(s, i)| s.map(|s| (s, i)))
{
let ssrc = info.ssrc;
let kind = info.kind;

let id = format!("RTCInboundRTP{}Stream_{}", capitalize(kind), ssrc);
let (
packets_received,
header_bytes_received,
bytes_received,
last_packet_received_timestamp,
nack_count,
remote_packets_sent,
remote_bytes_sent,
remote_reports_sent,
remote_round_trip_time,
remote_total_round_trip_time,
remote_round_trip_time_measurements,
) = (
stats.packets_received(),
stats.header_bytes_received(),
stats.payload_bytes_received(),
stats.last_packet_received_timestamp(),
stats.nacks_sent(),
stats.remote_packets_sent(),
stats.remote_bytes_sent(),
stats.remote_reports_sent(),
stats.remote_round_trip_time(),
stats.remote_total_round_trip_time(),
stats.remote_round_trip_time_measurements(),
);

collector.insert(
id.clone(),
crate::stats::StatsReportType::InboundRTP(InboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::InboundRTP,
id: id.clone(),
ssrc,
kind: kind.to_owned(),
for (stats, info) in
(stream_stats.into_iter().zip(track_infos)).filter_map(|(s, i)| s.map(|s| (s, i)))
{
let ssrc = info.ssrc;
let kind = info.kind;

let id = format!("RTCInboundRTP{}Stream_{}", capitalize(kind), ssrc);
let (
packets_received,
track_identifier: info.track_id,
mid: info.mid,
last_packet_received_timestamp,
header_bytes_received,
bytes_received,
last_packet_received_timestamp,
nack_count,
remote_packets_sent,
remote_bytes_sent,
remote_reports_sent,
remote_round_trip_time,
remote_total_round_trip_time,
remote_round_trip_time_measurements,
) = (
stats.packets_received(),
stats.header_bytes_received(),
stats.payload_bytes_received(),
stats.last_packet_received_timestamp(),
stats.nacks_sent(),
stats.remote_packets_sent(),
stats.remote_bytes_sent(),
stats.remote_reports_sent(),
stats.remote_round_trip_time(),
stats.remote_total_round_trip_time(),
stats.remote_round_trip_time_measurements(),
);

fir_count: (info.kind == "video").then(|| stats.firs_sent()),
pli_count: (info.kind == "video").then(|| stats.plis_sent()),
}),
);

let local_id = id;
let id = format!(
"RTCRemoteOutboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);
collector.insert(
id.clone(),
crate::stats::StatsReportType::RemoteOutboundRTP(RemoteOutboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::RemoteOutboundRTP,
id,
collector.insert(
id.clone(),
crate::stats::StatsReportType::InboundRTP(InboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::InboundRTP,
id: id.clone(),
ssrc,
kind: kind.to_owned(),
packets_received,
track_identifier: info.track_id,
mid: info.mid,
last_packet_received_timestamp,
header_bytes_received,
bytes_received,
nack_count,

fir_count: (info.kind == "video").then(|| stats.firs_sent()),
pli_count: (info.kind == "video").then(|| stats.plis_sent()),
}),
);

ssrc,
kind: kind.to_owned(),

packets_sent: remote_packets_sent as u64,
bytes_sent: remote_bytes_sent as u64,
local_id,
reports_sent: remote_reports_sent,
round_trip_time: remote_round_trip_time,
total_round_trip_time: remote_total_round_trip_time,
round_trip_time_measurements: remote_round_trip_time_measurements,
}),
);
let local_id = id;
let id = format!(
"RTCRemoteOutboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);
collector.insert(
id.clone(),
crate::stats::StatsReportType::RemoteOutboundRTP(RemoteOutboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::RemoteOutboundRTP,
id,

ssrc,
kind: kind.to_owned(),

packets_sent: remote_packets_sent as u64,
bytes_sent: remote_bytes_sent as u64,
local_id,
reports_sent: remote_reports_sent,
round_trip_time: remote_round_trip_time,
total_round_trip_time: remote_total_round_trip_time,
round_trip_time_measurements: remote_round_trip_time_measurements,
}),
);
}
}
}

@@ -1414,102 +1417,103 @@ impl PeerConnectionInternal {
}
}

let stream_stats = self
.stats_interceptor
.fetch_outbound_stats(track_infos.iter().map(|t| t.ssrc).collect())
.await;
if let Some(stats_interceptor) = self.stats_interceptor.upgrade() {
let stream_stats = stats_interceptor
.fetch_outbound_stats(track_infos.iter().map(|t| t.ssrc).collect())
.await;

for (stats, info) in stream_stats
.into_iter()
.zip(track_infos)
.filter_map(|(s, i)| s.map(|s| (s, i)))
{
// RTCOutboundRtpStreamStats
let id = format!(
"RTCOutboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);
let (
packets_sent,
bytes_sent,
header_bytes_sent,
nack_count,
remote_inbound_packets_received,
remote_inbound_packets_lost,
remote_rtt_ms,
remote_total_rtt_ms,
remote_rtt_measurements,
remote_fraction_lost,
) = (
stats.packets_sent(),
stats.payload_bytes_sent(),
stats.header_bytes_sent(),
stats.nacks_received(),
stats.remote_packets_received(),
stats.remote_total_lost(),
stats.remote_round_trip_time(),
stats.remote_total_round_trip_time(),
stats.remote_round_trip_time_measurements(),
stats.remote_fraction_lost(),
);

let TrackInfo {
mid,
ssrc,
rid,
kind,
track_id: track_identifier,
} = info;

collector.insert(
id.clone(),
crate::stats::StatsReportType::OutboundRTP(OutboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::OutboundRTP,
track_identifier,
id: id.clone(),
ssrc,
kind: kind.to_owned(),
for (stats, info) in stream_stats
.into_iter()
.zip(track_infos)
.filter_map(|(s, i)| s.map(|s| (s, i)))
{
// RTCOutboundRtpStreamStats
let id = format!(
"RTCOutboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);
let (
packets_sent,
mid,
rid,
header_bytes_sent,
bytes_sent,
header_bytes_sent,
nack_count,
remote_inbound_packets_received,
remote_inbound_packets_lost,
remote_rtt_ms,
remote_total_rtt_ms,
remote_rtt_measurements,
remote_fraction_lost,
) = (
stats.packets_sent(),
stats.payload_bytes_sent(),
stats.header_bytes_sent(),
stats.nacks_received(),
stats.remote_packets_received(),
stats.remote_total_lost(),
stats.remote_round_trip_time(),
stats.remote_total_round_trip_time(),
stats.remote_round_trip_time_measurements(),
stats.remote_fraction_lost(),
);

fir_count: (info.kind == "video").then(|| stats.firs_received()),
pli_count: (info.kind == "video").then(|| stats.plis_received()),
}),
);

let local_id = id;
let id = format!(
"RTCRemoteInboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);

collector.insert(
id.clone(),
StatsReportType::RemoteInboundRTP(RemoteInboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::RemoteInboundRTP,
id,
let TrackInfo {
mid,
ssrc,
kind: kind.to_owned(),

packets_received: remote_inbound_packets_received,
packets_lost: remote_inbound_packets_lost as i64,
rid,
kind,
track_id: track_identifier,
} = info;

collector.insert(
id.clone(),
crate::stats::StatsReportType::OutboundRTP(OutboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::OutboundRTP,
track_identifier,
id: id.clone(),
ssrc,
kind: kind.to_owned(),
packets_sent,
mid,
rid,
header_bytes_sent,
bytes_sent,
nack_count,

fir_count: (info.kind == "video").then(|| stats.firs_received()),
pli_count: (info.kind == "video").then(|| stats.plis_received()),
}),
);

local_id,
let local_id = id;
let id = format!(
"RTCRemoteInboundRTP{}Stream_{}",
capitalize(info.kind),
info.ssrc
);

round_trip_time: remote_rtt_ms,
total_round_trip_time: remote_total_rtt_ms,
fraction_lost: remote_fraction_lost.unwrap_or(0.0),
round_trip_time_measurements: remote_rtt_measurements,
}),
);
collector.insert(
id.clone(),
StatsReportType::RemoteInboundRTP(RemoteInboundRTPStats {
timestamp: Instant::now(),
stats_type: RTCStatsType::RemoteInboundRTP,
id,
ssrc,
kind: kind.to_owned(),

packets_received: remote_inbound_packets_received,
packets_lost: remote_inbound_packets_lost as i64,

local_id,

round_trip_time: remote_rtt_ms,
total_round_trip_time: remote_total_rtt_ms,
fraction_lost: remote_fraction_lost.unwrap_or(0.0),
round_trip_time_measurements: remote_rtt_measurements,
}),
);
}
}
}
}

0 comments on commit 8db3679

Please sign in to comment.