add statistics to Client, ServerSession, ServerConn, ServerStream (#556…

…) (#656)

client_format.go

@@ -1,6 +1,9 @@
 package gortsplib
 
 import (
+	"sync/atomic"
+	"time"
+
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
 
@@ -17,13 +20,29 @@ type clientFormat struct {
 	format      format.Format
 	onPacketRTP OnPacketRTPFunc
 
-	udpReorderer    *rtpreorderer.Reorderer       // play
-	tcpLossDetector *rtplossdetector.LossDetector // play
-	rtcpReceiver    *rtcpreceiver.RTCPReceiver    // play
-	rtcpSender      *rtcpsender.RTCPSender        // record or back channel
+	udpReorderer          *rtpreorderer.Reorderer       // play
+	tcpLossDetector       *rtplossdetector.LossDetector // play
+	rtcpReceiver          *rtcpreceiver.RTCPReceiver    // play
+	rtcpSender            *rtcpsender.RTCPSender        // record or back channel
+	writePacketRTPInQueue func([]byte) error
+	rtpPacketsReceived    *uint64
+	rtpPacketsSent        *uint64
+	rtpPacketsLost        *uint64
+}
+
+func (cf *clientFormat) initialize() {
+	cf.rtpPacketsReceived = new(uint64)
+	cf.rtpPacketsSent = new(uint64)
+	cf.rtpPacketsLost = new(uint64)
 }
 
 func (cf *clientFormat) start() {
+	if cf.cm.udpRTPListener != nil {
+		cf.writePacketRTPInQueue = cf.writePacketRTPInQueueUDP
+	} else {
+		cf.writePacketRTPInQueue = cf.writePacketRTPInQueueTCP
+	}
+
 	if cf.cm.c.state == clientStateRecord || cf.cm.media.IsBackChannel {
 		cf.rtcpSender = &rtcpsender.RTCPSender{
 			ClockRate: cf.format.ClockRate(),
@@ -72,40 +91,70 @@ func (cf *clientFormat) stop() {
 	}
 }
 
-func (cf *clientFormat) readRTPUDP(pkt *rtp.Packet) {
+func (cf *clientFormat) readPacketRTPUDP(pkt *rtp.Packet) {
 	packets, lost := cf.udpReorderer.Process(pkt)
 	if lost != 0 {
-		cf.cm.c.OnPacketLost(liberrors.ErrClientRTPPacketsLost{Lost: lost})
+		cf.onPacketRTPLost(lost)
 		// do not return
 	}
 
 	now := cf.cm.c.timeNow()
 
 	for _, pkt := range packets {
-		err := cf.rtcpReceiver.ProcessPacket(pkt, now, cf.format.PTSEqualsDTS(pkt))
-		if err != nil {
-			cf.cm.c.OnDecodeError(err)
-			continue
-		}
-
-		cf.onPacketRTP(pkt)
+		cf.handlePacketRTP(pkt, now)
 	}
 }
 
-func (cf *clientFormat) readRTPTCP(pkt *rtp.Packet) {
+func (cf *clientFormat) readPacketRTPTCP(pkt *rtp.Packet) {
 	lost := cf.tcpLossDetector.Process(pkt)
 	if lost != 0 {
-		cf.cm.c.OnPacketLost(liberrors.ErrClientRTPPacketsLost{Lost: lost})
+		cf.onPacketRTPLost(lost)
 		// do not return
 	}
 
 	now := cf.cm.c.timeNow()
 
-	err := cf.rtcpReceiver.ProcessPacket(pkt, now, cf.format.PTSEqualsDTS(pkt))
+	cf.handlePacketRTP(pkt, now)
+}
+
+func (cf *clientFormat) handlePacketRTP(pkt *rtp.Packet, now time.Time) {
+	err := cf.rtcpReceiver.ProcessPacketRTP(pkt, now, cf.format.PTSEqualsDTS(pkt))
 	if err != nil {
-		cf.cm.c.OnDecodeError(err)
+		cf.cm.onPacketRTPDecodeError(err)
 		return
 	}
 
+	atomic.AddUint64(cf.rtpPacketsReceived, 1)
+
 	cf.onPacketRTP(pkt)
 }
+
+func (cf *clientFormat) onPacketRTPLost(lost uint) {
+	atomic.AddUint64(cf.rtpPacketsLost, uint64(lost))
+	cf.cm.c.OnPacketLost(liberrors.ErrClientRTPPacketsLost{Lost: lost})
+}
+
+func (cf *clientFormat) writePacketRTPInQueueUDP(payload []byte) error {
+	err := cf.cm.udpRTPListener.write(payload)
+	if err != nil {
+		return err
+	}
+
+	atomic.AddUint64(cf.cm.bytesSent, uint64(len(payload)))
+	atomic.AddUint64(cf.rtpPacketsSent, 1)
+	return nil
+}
+
+func (cf *clientFormat) writePacketRTPInQueueTCP(payload []byte) error {
+	cf.cm.c.tcpFrame.Channel = cf.cm.tcpChannel
+	cf.cm.c.tcpFrame.Payload = payload
+	cf.cm.c.nconn.SetWriteDeadline(time.Now().Add(cf.cm.c.WriteTimeout))
+	err := cf.cm.c.conn.WriteInterleavedFrame(cf.cm.c.tcpFrame, cf.cm.c.tcpBuffer)
+	if err != nil {
+		return err
+	}
+
+	atomic.AddUint64(cf.cm.bytesSent, uint64(len(payload)))
+	atomic.AddUint64(cf.rtpPacketsSent, 1)
+	return nil
+}