1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
//! Statistics data carried by [`Event`][crate::Event].

use std::{
    collections::{HashMap, VecDeque},
    time::{Duration, Instant},
};

use crate::rtp::{Mid, Rid};

pub(crate) struct Stats {
    last_now: Instant,
    events: VecDeque<StatsEvent>,
    interval: Duration,
}

pub(crate) struct StatsSnapshot {
    pub peer_tx: u64,
    pub peer_rx: u64,
    pub tx: u64,
    pub rx: u64,
    pub ingress: HashMap<(Mid, Option<Rid>), MediaIngressStats>,
    pub egress: HashMap<(Mid, Option<Rid>), MediaEgressStats>,
    timestamp: Instant,
}

impl StatsSnapshot {
    pub(crate) fn new(timestamp: Instant) -> StatsSnapshot {
        StatsSnapshot {
            peer_rx: 0,
            peer_tx: 0,
            tx: 0,
            rx: 0,
            ingress: HashMap::new(),
            egress: HashMap::new(),
            timestamp,
        }
    }
}

// Output events

#[derive(Debug, Clone)]
pub(crate) enum StatsEvent {
    Peer(PeerStats),
    MediaEgress(MediaEgressStats),
    MediaIngress(MediaIngressStats),
}

/// An event representing the Peer statistics
///
/// This event is generated roughly every second
#[derive(Debug, Clone)]
pub struct PeerStats {
    /// Total bytes transmitted.
    pub peer_bytes_rx: u64,
    /// Total bytes received.
    pub peer_bytes_tx: u64,
    /// Total bytes transmitted, only counting media traffic (rtp payload).
    pub bytes_rx: u64,
    /// Total bytes received, only counting media traffic (rtp payload).
    pub bytes_tx: u64,
    /// Timestamp when this event was generated.
    pub timestamp: Instant,
}

/// An event carrying stats for every (mid, rid) in egress direction
///
/// note: when simulcast is disabled, `rid` is `None`
#[derive(Debug, Clone)]
pub struct MediaEgressStats {
    /// The identifier of the m-line these stats are for.
    pub mid: Mid,
    /// The Rid identifier in case of simulcast.
    pub rid: Option<Rid>,
    /// Total bytes sent, including retransmissions.
    ///
    /// Spec equivalent to [`RTCSentRtpStreamStats.bytesSent`][1].
    ///
    /// [1]: https://www.w3.org/TR/webrtc-stats/#dom-rtcsentrtpstreamstats-bytessent
    pub bytes: u64,
    /// Total number of rtp packets sent, including retransmissions
    ///
    /// Spec equivalent of [`RTCSentRtpStreamStats.packetsSent`][1].
    ///
    /// [1]: https://www.w3.org/TR/webrtc-stats/#dom-rtcsentrtpstreamstats-packetssent
    pub packets: u64,
    /// Number of firs received.
    pub firs: u64,
    /// Number of plis received.
    pub plis: u64,
    /// Number of nacks received.
    pub nacks: u64,
    /// Round-trip-time (ms) extracted from the last RTCP receiver report.
    pub rtt: Option<f32>,
    /// Fraction of packets lost averaged from the RTCP receiver reports received.
    /// `None` if no reports have been received since the last event
    pub loss: Option<f32>,
    /// Timestamp when this event was generated
    pub timestamp: Instant,
    // TODO
    // pub remote: RemoteIngressStats,
}

/// Stats as reported by the remote side (via RTCP ReceiverReports).
#[derive(Debug, Clone)]
pub struct RemoteIngressStats {
    /// Total bytes received.
    pub bytes_rx: u64,
}

/// An event carrying stats for every (mid, rid) in ingress direction
///
/// note: when simulcast is disabled, `rid` is `None`
#[derive(Debug, Clone)]
pub struct MediaIngressStats {
    /// The identifier of the m-line these stats are for.
    pub mid: Mid,
    /// The Rid identifier in case of simulcast.
    pub rid: Option<Rid>,
    /// Total bytes received, including retransmissions.
    pub bytes: u64,
    /// Total number of rtp packets received, including retransmissions.
    pub packets: u64,
    /// Number of firs sent.
    pub firs: u64,
    /// Number of plis sent.
    pub plis: u64,
    /// Number of nacks sent.
    pub nacks: u64,
    /// Round-trip-time (ms) extracted from the last RTCP XR DLRR report block.
    pub rtt: Option<f32>,
    /// Fraction of packets lost extracted from the last RTCP receiver report.
    pub loss: Option<f32>,
    /// Timestamp when this event was generated.
    pub timestamp: Instant,
    // TODO
    // pub remote: RemoteEgressStats,
}

/// Stats as reported by the remote side (via RTCP SenderReports).
#[derive(Debug, Clone)]
pub struct RemoteEgressStats {
    /// Total bytes transmitted.
    pub bytes_tx: u64,
}

impl Stats {
    /// Create a new stats instance
    ///
    /// The internal state is market with the current `Instant::now()`.
    /// This allows us to emit stats right away at the first upcoming timeout
    pub fn new(interval: Duration) -> Stats {
        Stats {
            // by starting with the current time we can generate stats right on first timeout
            last_now: Instant::now(),
            events: VecDeque::new(),
            interval,
        }
    }

    /// Returns true if we want to handle the timeout
    ///
    /// The caller can use this to conpute the snapshot only if needed, before calling [`Stats::do_handle_timeout`]
    pub fn wants_timeout(&mut self, now: Instant) -> bool {
        let min_step = self.last_now + self.interval;
        now >= min_step
    }

    /// Actually handles the timeout advancing the internal state and preparing the output
    pub fn do_handle_timeout(&mut self, snapshot: &mut StatsSnapshot) {
        // enqueue stas and timestampt them so they can be sent out

        let event = PeerStats {
            peer_bytes_rx: snapshot.peer_rx,
            peer_bytes_tx: snapshot.peer_tx,
            bytes_rx: snapshot.rx,
            bytes_tx: snapshot.tx,
            timestamp: snapshot.timestamp,
        };

        self.events.push_back(StatsEvent::Peer(event));

        for (_, event) in snapshot.ingress.drain() {
            self.events.push_back(StatsEvent::MediaIngress(event));
        }

        for (_, event) in snapshot.egress.drain() {
            self.events.push_back(StatsEvent::MediaEgress(event));
        }

        self.last_now = snapshot.timestamp;
    }

    /// Poll for the next time to call [`Stats::wants_timeout`] and [`Stats::do_handle_timeout`].
    ///
    /// NOTE: we only need Option<_> to conform to .soonest() (see caller)
    pub fn poll_timeout(&mut self) -> Option<Instant> {
        let last_now = self.last_now;
        Some(last_now + self.interval)
    }

    /// Return any events ready for delivery
    pub fn poll_output(&mut self) -> Option<StatsEvent> {
        self.events.pop_front()
    }
}