package/gortsplib
1
0
Fork 0
mirror of https://github.com/aler9/gortsplib synced 2025-10-27 01:00:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

add statistics to Client, ServerSession, ServerConn, ServerStream (#556) (#656)

Browse Source
This commit is contained in:
Alessandro Ros
2024-12-25 13:30:08 +01:00
committed by GitHub
parent 8c4a3ca018
commit 87c6d81053
34 changed files with 1246 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
client.go
View File

@@ -19,6 +19,8 @@ import (
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/bluenviron/gortsplib/v4/internal/rtcpreceiver"
"github.com/bluenviron/gortsplib/v4/internal/rtcpsender"
"github.com/bluenviron/gortsplib/v4/pkg/auth"
"github.com/bluenviron/gortsplib/v4/pkg/base"
"github.com/bluenviron/gortsplib/v4/pkg/bytecounter"
@@ -269,8 +271,10 @@ type Client struct {
// explicitly request back channels to the server.
RequestBackChannels bool
// pointer to a variable that stores received bytes.
// Deprecated: use Client.Stats()
BytesReceived *uint64
// pointer to a variable that stores sent bytes.
// Deprecated: use Client.Stats()
BytesSent *uint64
//
@@ -326,7 +330,7 @@ type Client struct {
effectiveTransport *Transport
backChannelSetupped bool
stdChannelSetupped bool
medias map[*description.Media]*clientMedia
setuppedMedias map[*description.Media]*clientMedia
tcpCallbackByChannel map[int]readFunc
lastRange *headers.Range
checkTimeoutTimer *time.Timer
@@ -341,6 +345,8 @@ type Client struct {
mustClose bool
tcpFrame *base.InterleavedFrame
tcpBuffer []byte
bytesReceived *uint64
bytesSent *uint64
// in
chOptions chan optionsReq
@@ -380,12 +386,6 @@ func (c *Client) Start(scheme string, host string) error {
if c.UserAgent == "" {
c.UserAgent = "gortsplib"
}
if c.BytesReceived == nil {
c.BytesReceived = new(uint64)
}
if c.BytesSent == nil {
c.BytesSent = new(uint64)
}
// system functions
if c.DialContext == nil {
@@ -454,6 +454,18 @@ func (c *Client) Start(scheme string, host string) error {
c.checkTimeoutTimer = emptyTimer()
c.keepalivePeriod = 30 * time.Second
c.keepaliveTimer = emptyTimer()
if c.BytesReceived != nil {
c.bytesReceived = c.BytesReceived
} else {
c.bytesReceived = new(uint64)
}
if c.BytesSent != nil {
c.bytesSent = c.BytesSent
} else {
c.bytesSent = new(uint64)
}
c.chOptions = make(chan optionsReq)
c.chDescribe = make(chan describeReq)
c.chAnnounce = make(chan announceReq)
@@ -739,7 +751,7 @@ func (c *Client) doClose() {
c.conn = nil
}
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
cm.close()
}
}
@@ -757,7 +769,7 @@ func (c *Client) reset() {
c.effectiveTransport = nil
c.backChannelSetupped = false
c.stdChannelSetupped = false
c.medias = nil
c.setuppedMedias = nil
c.tcpCallbackByChannel = nil
}
@@ -781,7 +793,7 @@ func (c *Client) trySwitchingProtocol() error {
prevConnURL := c.connURL
prevBaseURL := c.baseURL
prevMedias := c.medias
prevMedias := c.setuppedMedias
c.reset()
@@ -801,9 +813,9 @@ func (c *Client) trySwitchingProtocol() error {
return err
}
c.medias[i].onPacketRTCP = cm.onPacketRTCP
c.setuppedMedias[i].onPacketRTCP = cm.onPacketRTCP
for j, tr := range cm.formats {
c.medias[i].formats[j].onPacketRTP = tr.onPacketRTP
c.setuppedMedias[i].formats[j].onPacketRTP = tr.onPacketRTP
}
}
@@ -854,7 +866,7 @@ func (c *Client) startTransportRoutines() {
c.timeDecoder = rtptime.NewGlobalDecoder2()
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
cm.start()
}
@@ -894,7 +906,7 @@ func (c *Client) stopTransportRoutines() {
c.checkTimeoutTimer = emptyTimer()
c.keepaliveTimer = emptyTimer()
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
cm.stop()
}
@@ -935,7 +947,7 @@ func (c *Client) connOpen() error {
}
c.nconn = nconn
bc := bytecounter.New(c.nconn, c.BytesReceived, c.BytesSent)
bc := bytecounter.New(c.nconn, c.bytesReceived, c.bytesSent)
c.conn = conn.NewConn(bc)
c.reader = &clientReader{
c: c,
@@ -1021,7 +1033,7 @@ func (c *Client) do(req *base.Request, skipResponse bool) (*base.Response, error
}
func (c *Client) atLeastOneUDPPacketHasBeenReceived() bool {
for _, ct := range c.medias {
for _, ct := range c.setuppedMedias {
lft := atomic.LoadInt64(ct.udpRTPListener.lastPacketTime)
if lft != 0 {
return true
@@ -1037,7 +1049,7 @@ func (c *Client) atLeastOneUDPPacketHasBeenReceived() bool {
func (c *Client) isInUDPTimeout() bool {
now := c.timeNow()
for _, ct := range c.medias {
for _, ct := range c.setuppedMedias {
lft := time.Unix(atomic.LoadInt64(ct.udpRTPListener.lastPacketTime), 0)
if now.Sub(lft) < c.ReadTimeout {
return false
@@ -1347,9 +1359,10 @@ func (c *Client) doSetup(
}
cm := &clientMedia{
c: c,
onPacketRTCP: func(rtcp.Packet) {},
c: c,
media: medi,
}
cm.initialize()
if c.effectiveTransport == nil {
if c.connURL.Scheme == "rtsps" { // always use TCP if encrypted
@@ -1583,12 +1596,11 @@ func (c *Client) doSetup(
cm.tcpChannel = thRes.InterleavedIDs[0]
}
if c.medias == nil {
c.medias = make(map[*description.Media]*clientMedia)
if c.setuppedMedias == nil {
c.setuppedMedias = make(map[*description.Media]*clientMedia)
}
c.medias[medi] = cm
cm.setMedia(medi)
c.setuppedMedias[medi] = cm
c.baseURL = baseURL
c.effectiveTransport = &desiredTransport
@@ -1607,7 +1619,7 @@ func (c *Client) doSetup(
}
func (c *Client) isChannelPairInUse(channel int) bool {
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
if (cm.tcpChannel+1) == channel || cm.tcpChannel == channel || cm.tcpChannel == (channel+1) {
return true
}
@@ -1712,7 +1724,7 @@ func (c *Client) doPlay(ra *headers.Range) (*base.Response, error) {
// don't do this with multicast, otherwise the RTP packet is going to be broadcasted
// to all listeners, including us, messing up the stream.
if *c.effectiveTransport == TransportUDP {
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
byts, _ := (&rtp.Packet{Header: rtp.Header{Version: 2}}).Marshal()
cm.udpRTPListener.write(byts) //nolint:errcheck
@@ -1852,9 +1864,9 @@ func (c *Client) Seek(ra *headers.Range) (*base.Response, error) {
return c.Play(ra)
}
// OnPacketRTPAny sets the callback that is called when a RTP packet is read from any setupped media.
// OnPacketRTPAny sets a callback that is called when a RTP packet is read from any setupped media.
func (c *Client) OnPacketRTPAny(cb OnPacketRTPAnyFunc) {
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
cmedia := cm.media
for _, forma := range cm.media.Formats {
c.OnPacketRTP(cm.media, forma, func(pkt *rtp.Packet) {
@@ -1864,9 +1876,9 @@ func (c *Client) OnPacketRTPAny(cb OnPacketRTPAnyFunc) {
}
}
// OnPacketRTCPAny sets the callback that is called when a RTCP packet is read from any setupped media.
// OnPacketRTCPAny sets a callback that is called when a RTCP packet is read from any setupped media.
func (c *Client) OnPacketRTCPAny(cb OnPacketRTCPAnyFunc) {
for _, cm := range c.medias {
for _, cm := range c.setuppedMedias {
cmedia := cm.media
c.OnPacketRTCP(cm.media, func(pkt rtcp.Packet) {
cb(cmedia, pkt)
@@ -1874,16 +1886,16 @@ func (c *Client) OnPacketRTCPAny(cb OnPacketRTCPAnyFunc) {
}
}
// OnPacketRTP sets the callback that is called when a RTP packet is read.
// OnPacketRTP sets a callback that is called when a RTP packet is read.
func (c *Client) OnPacketRTP(medi *description.Media, forma format.Format, cb OnPacketRTPFunc) {
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
ct := cm.formats[forma.PayloadType()]
ct.onPacketRTP = cb
}
// OnPacketRTCP sets the callback that is called when a RTCP packet is read.
// OnPacketRTCP sets a callback that is called when a RTCP packet is read.
func (c *Client) OnPacketRTCP(medi *description.Media, cb OnPacketRTCPFunc) {
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
cm.onPacketRTCP = cb
}
@@ -1908,13 +1920,13 @@ func (c *Client) WritePacketRTPWithNTP(medi *description.Media, pkt *rtp.Packet,
default:
}
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
cf := cm.formats[pkt.PayloadType]
cf.rtcpSender.ProcessPacket(pkt, ntp, cf.format.PTSEqualsDTS(pkt))
cf.rtcpSender.ProcessPacketRTP(pkt, ntp, cf.format.PTSEqualsDTS(pkt))
ok := c.writer.push(func() error {
return cm.writePacketRTPInQueue(byts)
return cf.writePacketRTPInQueue(byts)
})
if !ok {
return liberrors.ErrClientWriteQueueFull{}
@@ -1936,7 +1948,7 @@ func (c *Client) WritePacketRTCP(medi *description.Media, pkt rtcp.Packet) error
default:
}
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
ok := c.writer.push(func() error {
return cm.writePacketRTCPInQueue(byts)
@@ -1953,7 +1965,7 @@ func (c *Client) WritePacketRTCP(medi *description.Media, pkt rtcp.Packet) error
//
// Deprecated: replaced by PacketPTS2.
func (c *Client) PacketPTS(medi *description.Media, pkt *rtp.Packet) (time.Duration, bool) {
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
ct := cm.formats[pkt.PayloadType]
v, ok := c.timeDecoder.Decode(ct.format, pkt)
@@ -1967,7 +1979,7 @@ func (c *Client) PacketPTS(medi *description.Media, pkt *rtp.Packet) (time.Durat
// PacketPTS2 returns the PTS of an incoming RTP packet.
// It is computed by decoding the packet timestamp and sychronizing it with other tracks.
func (c *Client) PacketPTS2(medi *description.Media, pkt *rtp.Packet) (int64, bool) {
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
ct := cm.formats[pkt.PayloadType]
return c.timeDecoder.Decode(ct.format, pkt)
}
@@ -1975,7 +1987,194 @@ func (c *Client) PacketPTS2(medi *description.Media, pkt *rtp.Packet) (int64, bo
// PacketNTP returns the NTP timestamp of an incoming RTP packet.
// The NTP timestamp is computed from RTCP sender reports.
func (c *Client) PacketNTP(medi *description.Media, pkt *rtp.Packet) (time.Time, bool) {
cm := c.medias[medi]
cm := c.setuppedMedias[medi]
ct := cm.formats[pkt.PayloadType]
return ct.rtcpReceiver.PacketNTP(pkt.Timestamp)
}
// Stats returns client statistics.
func (c *Client) Stats() *ClientStats {
return &ClientStats{
Conn: StatsConn{
BytesReceived: atomic.LoadUint64(c.bytesReceived),
BytesSent: atomic.LoadUint64(c.bytesSent),
},
Session: StatsSession{
BytesReceived: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.bytesReceived)
}
return v
}(),
BytesSent: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.bytesSent)
}
return v
}(),
RTPPacketsReceived: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
for _, f := range sm.formats {
v += atomic.LoadUint64(f.rtpPacketsReceived)
}
}
return v
}(),
RTPPacketsSent: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
for _, f := range sm.formats {
v += atomic.LoadUint64(f.rtpPacketsSent)
}
}
return v
}(),
RTPPacketsLost: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
for _, f := range sm.formats {
v += atomic.LoadUint64(f.rtpPacketsLost)
}
}
return v
}(),
RTPPacketsInError: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.rtpPacketsInError)
}
return v
}(),
RTPJitter: func() float64 {
v := float64(0)
n := float64(0)
for _, sm := range c.setuppedMedias {
for _, fo := range sm.formats {
if fo.rtcpReceiver != nil {
stats := fo.rtcpReceiver.Stats()
if stats != nil {
v += stats.Jitter
n++
}
}
}
}
return v / n
}(),
RTCPPacketsReceived: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.rtcpPacketsReceived)
}
return v
}(),
RTCPPacketsSent: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.rtcpPacketsSent)
}
return v
}(),
RTCPPacketsInError: func() uint64 {
v := uint64(0)
for _, sm := range c.setuppedMedias {
v += atomic.LoadUint64(sm.rtcpPacketsInError)
}
return v
}(),
Medias: func() map[*description.Media]StatsSessionMedia { //nolint:dupl
ret := make(map[*description.Media]StatsSessionMedia, len(c.setuppedMedias))
for med, sm := range c.setuppedMedias {
ret[med] = StatsSessionMedia{
BytesReceived: atomic.LoadUint64(sm.bytesReceived),
BytesSent: atomic.LoadUint64(sm.bytesSent),
RTPPacketsInError: atomic.LoadUint64(sm.rtpPacketsInError),
RTCPPacketsReceived: atomic.LoadUint64(sm.rtcpPacketsReceived),
RTCPPacketsSent: atomic.LoadUint64(sm.rtcpPacketsSent),
RTCPPacketsInError: atomic.LoadUint64(sm.rtcpPacketsInError),
Formats: func() map[format.Format]StatsSessionFormat {
ret := make(map[format.Format]StatsSessionFormat, len(sm.formats))
for _, fo := range sm.formats {
recvStats := func() *rtcpreceiver.Stats {
if fo.rtcpReceiver != nil {
return fo.rtcpReceiver.Stats()
}
return nil
}()
sentStats := func() *rtcpsender.Stats {
if fo.rtcpSender != nil {
return fo.rtcpSender.Stats()
}
return nil
}()
ret[fo.format] = StatsSessionFormat{ //nolint:dupl
RTPPacketsReceived: atomic.LoadUint64(fo.rtpPacketsReceived),
RTPPacketsSent: atomic.LoadUint64(fo.rtpPacketsSent),
RTPPacketsLost: atomic.LoadUint64(fo.rtpPacketsLost),
LocalSSRC: func() uint32 {
if fo.rtcpReceiver != nil {
return *fo.rtcpReceiver.LocalSSRC
}
if sentStats != nil {
return sentStats.LocalSSRC
}
return 0
}(),
RemoteSSRC: func() uint32 {
if recvStats != nil {
return recvStats.RemoteSSRC
}
return 0
}(),
RTPPacketsLastSequenceNumber: func() uint16 {
if recvStats != nil {
return recvStats.LastSequenceNumber
}
if sentStats != nil {
return sentStats.LastSequenceNumber
}
return 0
}(),
RTPPacketsLastRTP: func() uint32 {
if recvStats != nil {
return recvStats.LastRTP
}
if sentStats != nil {
return sentStats.LastRTP
}
return 0
}(),
RTPPacketsLastNTP: func() time.Time {
if recvStats != nil {
return recvStats.LastNTP
}
if sentStats != nil {
return sentStats.LastNTP
}
return time.Time{}
}(),
RTPJitter: func() float64 {
if recvStats != nil {
return recvStats.Jitter
}
return 0
}(),
}
}
return ret
}(),
}
}
return ret
}(),
},
}
}