expose number of lost packets without passing through an error (#735)

This commit is contained in:
Alessandro Ros
2025-03-24 16:39:55 +01:00
committed by GitHub
parent 1b127d70bb
commit c0c275e6a6
13 changed files with 68 additions and 28 deletions

View File

@@ -217,8 +217,13 @@ type ClientOnResponseFunc func(*base.Response)
type ClientOnTransportSwitchFunc func(err error)
// ClientOnPacketLostFunc is the prototype of Client.OnPacketLost.
//
// Deprecated: replaced by ClientOnPacketLost2Func
type ClientOnPacketLostFunc func(err error)
// ClientOnPacketLost2Func is the prototype of Client.OnPacketLost2.
type ClientOnPacketLost2Func func(lost uint64)
// ClientOnDecodeErrorFunc is the prototype of Client.OnDecodeError.
type ClientOnDecodeErrorFunc func(err error)
@@ -276,9 +281,11 @@ 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
@@ -306,7 +313,11 @@ type Client struct {
// called when the transport protocol changes.
OnTransportSwitch ClientOnTransportSwitchFunc
// called when the client detects lost packets.
//
// Deprecated: replaced by OnPacketLost2.
OnPacketLost ClientOnPacketLostFunc
// called when the client detects lost packets.
OnPacketLost2 ClientOnPacketLost2Func
// called when a non-fatal decode error occurs.
OnDecodeError ClientOnDecodeErrorFunc
@@ -423,9 +434,21 @@ func (c *Client) Start(scheme string, host string) error {
log.Println(err.Error())
}
}
if c.OnPacketLost == nil {
c.OnPacketLost = func(err error) {
log.Println(err.Error())
if c.OnPacketLost != nil {
c.OnPacketLost2 = func(lost uint64) {
c.OnPacketLost(liberrors.ErrClientRTPPacketsLost{Lost: uint(lost)}) //nolint:staticcheck
}
}
if c.OnPacketLost2 == nil {
c.OnPacketLost2 = func(lost uint64) {
log.Printf("%d RTP %s lost",
lost,
func() string {
if lost == 1 {
return "packet"
}
return "packets"
}())
}
}
if c.OnDecodeError == nil {

View File

@@ -12,7 +12,6 @@ import (
"github.com/bluenviron/gortsplib/v4/internal/rtplossdetector"
"github.com/bluenviron/gortsplib/v4/internal/rtpreorderer"
"github.com/bluenviron/gortsplib/v4/pkg/format"
"github.com/bluenviron/gortsplib/v4/pkg/liberrors"
)
type clientFormat struct {
@@ -129,9 +128,9 @@ func (cf *clientFormat) handlePacketRTP(pkt *rtp.Packet, now time.Time) {
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) onPacketRTPLost(lost uint64) {
atomic.AddUint64(cf.rtpPacketsLost, lost)
cf.cm.c.OnPacketLost2(lost)
}
func (cf *clientFormat) writePacketRTPInQueueUDP(payload []byte) error {

View File

@@ -13,7 +13,7 @@ type LossDetector struct {
// Process processes a RTP packet.
// It returns the number of lost packets.
func (r *LossDetector) Process(pkt *rtp.Packet) uint {
func (r *LossDetector) Process(pkt *rtp.Packet) uint64 {
if !r.initialized {
r.initialized = true
r.expectedSeqNum = pkt.SequenceNumber + 1
@@ -23,7 +23,7 @@ func (r *LossDetector) Process(pkt *rtp.Packet) uint {
if pkt.SequenceNumber != r.expectedSeqNum {
diff := pkt.SequenceNumber - r.expectedSeqNum
r.expectedSeqNum = pkt.SequenceNumber + 1
return uint(diff)
return uint64(diff)
}
r.expectedSeqNum = pkt.SequenceNumber + 1

View File

@@ -15,19 +15,19 @@ func TestLossDetector(t *testing.T) {
SequenceNumber: 65530,
},
})
require.Equal(t, uint(0), c)
require.Equal(t, uint64(0), c)
c = d.Process(&rtp.Packet{
Header: rtp.Header{
SequenceNumber: 65531,
},
})
require.Equal(t, uint(0), c)
require.Equal(t, uint64(0), c)
c = d.Process(&rtp.Packet{
Header: rtp.Header{
SequenceNumber: 65535,
},
})
require.Equal(t, uint(3), c)
require.Equal(t, uint64(3), c)
}

View File

@@ -27,7 +27,7 @@ func (r *Reorderer) Initialize() {
// Process processes a RTP packet.
// It returns a sequence of ordered packets and the number of lost packets.
func (r *Reorderer) Process(pkt *rtp.Packet) ([]*rtp.Packet, uint) {
func (r *Reorderer) Process(pkt *rtp.Packet) ([]*rtp.Packet, uint64) {
if !r.initialized {
r.initialized = true
r.expectedSeqNum = pkt.SequenceNumber + 1
@@ -86,7 +86,7 @@ func (r *Reorderer) Process(pkt *rtp.Packet) ([]*rtp.Packet, uint) {
ret[pos] = pkt
r.expectedSeqNum = pkt.SequenceNumber + 1
return ret, uint(int(relPos) - n + 1)
return ret, uint64(int(relPos) - n + 1)
}
// there's a missing packet

View File

@@ -164,7 +164,7 @@ func TestReorder(t *testing.T) {
for _, entry := range sequence {
out, missing := r.Process(entry.in)
require.Equal(t, entry.out, out)
require.Equal(t, uint(0), missing)
require.Equal(t, uint64(0), missing)
}
}
@@ -173,7 +173,7 @@ func TestBufferIsFull(t *testing.T) {
r.Initialize()
r.absPos = 25
sn := uint16(1564)
toMiss := uint(34)
toMiss := uint64(34)
out, missing := r.Process(&rtp.Packet{
Header: rtp.Header{
@@ -185,19 +185,19 @@ func TestBufferIsFull(t *testing.T) {
SequenceNumber: sn,
},
}}, out)
require.Equal(t, uint(0), missing)
require.Equal(t, uint64(0), missing)
sn++
var expected []*rtp.Packet
for i := uint(0); i < 64-toMiss; i++ {
for i := uint64(0); i < 64-toMiss; i++ {
out, missing = r.Process(&rtp.Packet{
Header: rtp.Header{
SequenceNumber: sn + uint16(toMiss),
},
})
require.Equal(t, []*rtp.Packet(nil), out)
require.Equal(t, uint(0), missing)
require.Equal(t, uint64(0), missing)
expected = append(expected, &rtp.Packet{
Header: rtp.Header{
@@ -242,7 +242,7 @@ func TestReset(t *testing.T) {
},
})
require.Equal(t, []*rtp.Packet(nil), out)
require.Equal(t, uint(0), missing)
require.Equal(t, uint64(0), missing)
sn++
}
@@ -256,5 +256,5 @@ func TestReset(t *testing.T) {
SequenceNumber: sn,
},
}}, out)
require.Equal(t, uint(0), missing)
require.Equal(t, uint64(0), missing)
}

View File

@@ -17,6 +17,7 @@ type Opus struct {
PayloadTyp uint8
ChannelCount int
//
// Deprecated: replaced by ChannelCount.
IsStereo bool
}

View File

@@ -250,6 +250,8 @@ func (e ErrClientWriteQueueFull) Error() string {
}
// ErrClientRTPPacketsLost is an error that can be returned by a client.
//
// Deprecated: will be removed in next version.
type ErrClientRTPPacketsLost struct {
Lost uint
}

View File

@@ -239,6 +239,8 @@ func (e ErrServerUnexpectedResponse) Error() string {
type ErrServerWriteQueueFull = ErrClientWriteQueueFull
// ErrServerRTPPacketsLost is an error that can be returned by a server.
//
// Deprecated: will be removed in next version.
type ErrServerRTPPacketsLost = ErrClientRTPPacketsLost
// ErrServerRTPPacketUnknownPayloadType is an error that can be returned by a server.

View File

@@ -19,5 +19,5 @@ func New() *LossDetector {
// Process processes a RTP packet.
// It returns the number of lost packets.
func (r *LossDetector) Process(pkt *rtp.Packet) uint {
return (*rtplossdetector.LossDetector)(r).Process(pkt)
return uint((*rtplossdetector.LossDetector)(r).Process(pkt))
}

View File

@@ -23,5 +23,6 @@ func New() *Reorderer {
// Process processes a RTP packet.
// It returns a sequence of ordered packets and the number of lost packets.
func (r *Reorderer) Process(pkt *rtp.Packet) ([]*rtp.Packet, uint) {
return (*rtpreorderer.Reorderer)(r).Process(pkt)
v1, v2 := (*rtpreorderer.Reorderer)(r).Process(pkt)
return v1, uint(v2)
}

View File

@@ -195,7 +195,11 @@ type ServerHandlerOnSetParameter interface {
// ServerHandlerOnPacketLostCtx is the context of OnPacketLost.
type ServerHandlerOnPacketLostCtx struct {
Session *ServerSession
Error error
Lost uint64
//
// Deprecated: replaced by Lost
Error error
}
// ServerHandlerOnPacketLost can be implemented by a ServerHandler.

View File

@@ -112,16 +112,24 @@ func (sf *serverSessionFormat) handlePacketRTP(pkt *rtp.Packet, now time.Time) {
sf.onPacketRTP(pkt)
}
func (sf *serverSessionFormat) onPacketRTPLost(lost uint) {
atomic.AddUint64(sf.rtpPacketsLost, uint64(lost))
func (sf *serverSessionFormat) onPacketRTPLost(lost uint64) {
atomic.AddUint64(sf.rtpPacketsLost, lost)
if h, ok := sf.sm.ss.s.Handler.(ServerHandlerOnPacketLost); ok {
h.OnPacketLost(&ServerHandlerOnPacketLostCtx{
Session: sf.sm.ss,
Error: liberrors.ErrServerRTPPacketsLost{Lost: lost},
Lost: lost,
Error: liberrors.ErrServerRTPPacketsLost{Lost: uint(lost)}, //nolint:staticcheck
})
} else {
log.Println(liberrors.ErrServerRTPPacketsLost{Lost: lost}.Error())
log.Printf("%d RTP %s lost",
lost,
func() string {
if lost == 1 {
return "packet"
}
return "packets"
}())
}
}