package/gortsplib
1
0
Fork 0
mirror of https://github.com/aler9/gortsplib synced 2025-09-29 04:22:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

allow to set NTP of outgoing packets in both client and server (#148)

Browse Source
This commit is contained in:
Alessandro Ros
2022-11-14 16:55:40 +01:00
committed by GitHub
parent ae0df8d4c9
commit 7951b2e4af
19 changed files with 95 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
client.go
View File

@@ -163,7 +163,6 @@ type clientRes struct {
type ClientOnPacketRTPCtx struct {
TrackID int
Packet *rtp.Packet
PTSEqualsDTS bool
}
// ClientOnPacketRTCPCtx is the context of a RTCP packet.
@@ -816,7 +815,6 @@ func (c *Client) runReader() {
c.OnPacketRTP(&ClientOnPacketRTPCtx{
TrackID: track.id,
Packet: pkt,
PTSEqualsDTS: ptsEqualsDTS(track.track, pkt),
})
} else {
if len(payload) > maxPacketSize {
@@ -1872,8 +1870,15 @@ func (c *Client) runWriter() {
}
}
// WritePacketRTP writes a RTP packet.
func (c *Client) WritePacketRTP(trackID int, pkt *rtp.Packet, ptsEqualsDTS bool) error {
// WritePacketRTP writes a RTP packet to all the readers of the stream.
func (c *Client) WritePacketRTP(trackID int, pkt *rtp.Packet) error {
return c.WritePacketRTPWithNTP(trackID, pkt, time.Now())
}
// WritePacketRTPWithNTP writes a RTP packet.
// ntp is the absolute time of the packet, and is needed to generate RTCP sender reports
// that allows the receiver to reconstruct the absolute time of the packet.
func (c *Client) WritePacketRTPWithNTP(trackID int, pkt *rtp.Packet, ntp time.Time) error {
c.writeMutex.RLock()
defer c.writeMutex.RUnlock()
@@ -1895,7 +1900,7 @@ func (c *Client) WritePacketRTP(trackID int, pkt *rtp.Packet, ptsEqualsDTS bool)
t := c.tracks[trackID]
if t.rtcpSender != nil {
t.rtcpSender.ProcessPacketRTP(time.Now(), pkt, ptsEqualsDTS)
t.rtcpSender.ProcessPacketRTP(ntp, pkt, ptsEqualsDTS(c.tracks[trackID].track, pkt))
}
c.writeBuffer.Push(trackTypePayload{

30
client_publish_test.go
View File

@@ -246,14 +246,14 @@ func TestClientPublishSerial(t *testing.T) {
c.Wait()
}()
err = c.WritePacketRTP(0, &testRTPPacket, true)
err = c.WritePacketRTP(0, &testRTPPacket)
require.NoError(t, err)
<-recvDone
c.Close()
<-done
err = c.WritePacketRTP(0, &testRTPPacket, true)
err = c.WritePacketRTP(0, &testRTPPacket)
require.Error(t, err)
})
}
@@ -403,7 +403,7 @@ func TestClientPublishParallel(t *testing.T) {
defer t.Stop()
for range t.C {
err := c.WritePacketRTP(0, &testRTPPacket, true)
err := c.WritePacketRTP(0, &testRTPPacket)
if err != nil {
return
}
@@ -552,7 +552,7 @@ func TestClientPublishPauseSerial(t *testing.T) {
require.NoError(t, err)
defer c.Close()
err = c.WritePacketRTP(0, &testRTPPacket, true)
err = c.WritePacketRTP(0, &testRTPPacket)
require.NoError(t, err)
_, err = c.Pause()
@@ -561,7 +561,7 @@ func TestClientPublishPauseSerial(t *testing.T) {
_, err = c.Record()
require.NoError(t, err)
err = c.WritePacketRTP(0, &testRTPPacket, true)
err = c.WritePacketRTP(0, &testRTPPacket)
require.NoError(t, err)
})
}
@@ -693,7 +693,7 @@ func TestClientPublishPauseParallel(t *testing.T) {
defer t.Stop()
for range t.C {
err := c.WritePacketRTP(0, &testRTPPacket, true)
err := c.WritePacketRTP(0, &testRTPPacket)
if err != nil {
return
}
@@ -829,7 +829,7 @@ func TestClientPublishAutomaticProtocol(t *testing.T) {
require.NoError(t, err)
defer c.Close()
err = c.WritePacketRTP(0, &testRTPPacket, true)
err = c.WritePacketRTP(0, &testRTPPacket)
require.NoError(t, err)
}
@@ -952,7 +952,7 @@ func TestClientPublishRTCPReport(t *testing.T) {
NTPTime: packets[0].(*rtcp.SenderReport).NTPTime,
RTPTime: packets[0].(*rtcp.SenderReport).RTPTime,
PacketCount: 2,
OctetCount: 8,
OctetCount: 2,
}, packets[0])
close(reportReceived)
@@ -988,11 +988,17 @@ func TestClientPublishRTCPReport(t *testing.T) {
require.NoError(t, err)
defer c.Close()
err = c.WritePacketRTP(0, &testRTPPacket, true)
require.NoError(t, err)
err = c.WritePacketRTP(0, &testRTPPacket, true)
for i := 0; i < 2; i++ {
err = c.WritePacketRTP(0, &rtp.Packet{
Header: rtp.Header{
Version: 2,
PayloadType: 96,
SSRC: 0x38F27A2F,
},
Payload: []byte{0x05}, // IDR
})
require.NoError(t, err)
}
<-reportReceived
})

1
clientudpl.go
View File

@@ -220,7 +220,6 @@ func (u *clientUDPListener) processPlayRTP(now time.Time, payload []byte) {
u.c.OnPacketRTP(&ClientOnPacketRTPCtx{
TrackID: u.ct.id,
Packet: pkt,
PTSEqualsDTS: ptsEqualsDTS,
})
}
}

2
examples/client-publish-codec-h264/main.go
View File

@@ -57,7 +57,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-codec-mpeg4audio/main.go
View File

@@ -65,7 +65,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-codec-opus/main.go
View File

@@ -59,7 +59,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-codec-pcma/main.go
View File

@@ -55,7 +55,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-codec-pcmu/main.go
View File

@@ -55,7 +55,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-options/main.go
View File

@@ -67,7 +67,7 @@ func main() {
}
// route RTP packet to the server
err = c.WritePacketRTP(0, &pkt, true)
err = c.WritePacketRTP(0, &pkt)
if err != nil {
panic(err)
}

2
examples/client-publish-pause/main.go
View File

@@ -61,7 +61,7 @@ func main() {
}
// route RTP packet to the server
c.WritePacketRTP(0, &pkt, true)
c.WritePacketRTP(0, &pkt)
// read another RTP packet from source
n, _, err = pc.ReadFrom(buf)

2
examples/client-read-republish/main.go
View File

@@ -46,7 +46,7 @@ func main() {
// called when a RTP packet arrives
reader.OnPacketRTP = func(ctx *gortsplib.ClientOnPacketRTPCtx) {
publisher.WritePacketRTP(ctx.TrackID, ctx.Packet, ctx.PTSEqualsDTS)
publisher.WritePacketRTP(ctx.TrackID, ctx.Packet)
}
// setup and read all tracks

2
examples/server-tls/main.go
View File

@@ -134,7 +134,7 @@ func (sh *serverHandler) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx)
// if we are the publisher, route the RTP packet to all readers
if ctx.Session == sh.publisher {
sh.stream.WritePacketRTP(ctx.TrackID, ctx.Packet, ctx.PTSEqualsDTS)
sh.stream.WritePacketRTP(ctx.TrackID, ctx.Packet)
}
}

2
examples/server/main.go
View File

@@ -133,7 +133,7 @@ func (sh *serverHandler) OnPacketRTP(ctx *gortsplib.ServerHandlerOnPacketRTPCtx)
// if we are the publisher, route the RTP packet to all readers
if ctx.Session == sh.publisher {
sh.stream.WritePacketRTP(ctx.TrackID, ctx.Packet, ctx.PTSEqualsDTS)
sh.stream.WritePacketRTP(ctx.TrackID, ctx.Packet)
}
}

2
internal/highleveltests/server_test.go
View File

@@ -405,7 +405,7 @@ func TestServerPublishRead(t *testing.T) {
defer mutex.Unlock()
if ctx.Session == publisher {
stream.WritePacketRTP(ctx.TrackID, ctx.Packet, true)
stream.WritePacketRTP(ctx.TrackID, ctx.Packet)
}
},
},

46
server_read_test.go
View File

@@ -415,7 +415,7 @@ func TestServerRead(t *testing.T) {
go func() {
time.Sleep(1 * time.Second)
stream.WritePacketRTCP(0, &testRTCPPacket)
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
}()
return &base.Response{
@@ -803,8 +803,16 @@ func TestServerReadRTCPReport(t *testing.T) {
require.NoError(t, err)
require.Equal(t, base.StatusOK, res.StatusCode)
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket, true)
for i := 0; i < 2; i++ {
stream.WritePacketRTP(0, &rtp.Packet{
Header: rtp.Header{
Version: 2,
PayloadType: 96,
SSRC: 0x38F27A2F,
},
Payload: []byte{0x05}, // IDR
})
}
var buf []byte
@@ -833,7 +841,7 @@ func TestServerReadRTCPReport(t *testing.T) {
NTPTime: packets[0].(*rtcp.SenderReport).NTPTime,
RTPTime: packets[0].(*rtcp.SenderReport).RTPTime,
PacketCount: 2,
OctetCount: 8,
OctetCount: 2,
}, packets[0])
res, err = writeReqReadRes(conn, base.Request{
@@ -931,7 +939,7 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
go func() {
defer close(writerDone)
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
t := time.NewTicker(50 * time.Millisecond)
defer t.Stop()
@@ -939,7 +947,7 @@ func TestServerReadTCPResponseBeforeFrames(t *testing.T) {
for {
select {
case <-t.C:
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
case <-writerTerminate:
return
}
@@ -1128,7 +1136,7 @@ func TestServerReadPlayPausePlay(t *testing.T) {
for {
select {
case <-t.C:
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
case <-writerTerminate:
return
}
@@ -1252,7 +1260,7 @@ func TestServerReadPlayPausePause(t *testing.T) {
for {
select {
case <-t.C:
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
case <-writerTerminate:
return
}
@@ -1708,8 +1716,8 @@ func TestServerReadPartialTracks(t *testing.T) {
onPlay: func(ctx *ServerHandlerOnPlayCtx) (*base.Response, error) {
go func() {
time.Sleep(1 * time.Second)
stream.WritePacketRTP(0, &testRTPPacket, true)
stream.WritePacketRTP(1, &testRTPPacket, true)
stream.WritePacketRTP(0, &testRTPPacket)
stream.WritePacketRTP(1, &testRTPPacket)
}()
return &base.Response{
@@ -1864,11 +1872,15 @@ func TestServerReadAdditionalInfos(t *testing.T) {
return &ri, ssrcs
}
track := &TrackH264{
PayloadType: 96,
SPS: []byte{0x01, 0x02, 0x03, 0x04},
PPS: []byte{0x01, 0x02, 0x03, 0x04},
track := &TrackGeneric{
Media: "application",
Payloads: []TrackGenericPayload{{
Type: 96,
RTPMap: "private/90000",
}},
}
err := track.Init()
require.NoError(t, err)
stream := NewServerStream(Tracks{track, track})
defer stream.Close()
@@ -1889,7 +1901,7 @@ func TestServerReadAdditionalInfos(t *testing.T) {
RTSPAddress: "localhost:8554",
}
err := s.Start()
err = s.Start()
require.NoError(t, err)
defer s.Close()
@@ -1902,7 +1914,7 @@ func TestServerReadAdditionalInfos(t *testing.T) {
SSRC: 96342362,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
}, true)
})
rtpInfo, ssrcs := getInfos()
require.Equal(t, &headers.RTPInfo{
@@ -1936,7 +1948,7 @@ func TestServerReadAdditionalInfos(t *testing.T) {
SSRC: 536474323,
},
Payload: []byte{0x01, 0x02, 0x03, 0x04},
}, true)
})
rtpInfo, ssrcs = getInfos()
require.Equal(t, &headers.RTPInfo{

1
serverconn.go
View File

@@ -283,7 +283,6 @@ func (sc *ServerConn) readFuncTCP(readRequest chan readReq) error {
Session: sc.session,
TrackID: track.id,
Packet: pkt,
PTSEqualsDTS: ptsEqualsDTS(track.track, pkt),
})
}
} else {

1
serverhandler.go
View File

@@ -202,7 +202,6 @@ type ServerHandlerOnPacketRTPCtx struct {
Session *ServerSession
TrackID int
Packet *rtp.Packet
PTSEqualsDTS bool
}
// ServerHandlerOnPacketRTP can be implemented by a ServerHandler.

15
serverstream.go
View File

@@ -232,7 +232,14 @@ func (st *ServerStream) readerSetInactive(ss *ServerSession) {
}
// WritePacketRTP writes a RTP packet to all the readers of the stream.
func (st *ServerStream) WritePacketRTP(trackID int, pkt *rtp.Packet, ptsEqualsDTS bool) {
func (st *ServerStream) WritePacketRTP(trackID int, pkt *rtp.Packet) {
st.WritePacketRTPWithNTP(trackID, pkt, time.Now())
}
// WritePacketRTPWithNTP writes a RTP packet to all the readers of the stream.
// ntp is the absolute time of the packet, and is needed to generate RTCP sender reports
// that allows the receiver to reconstruct the absolute time of the packet.
func (st *ServerStream) WritePacketRTPWithNTP(trackID int, pkt *rtp.Packet, ntp time.Time) {
byts := make([]byte, maxPacketSize)
n, err := pkt.MarshalTo(byts)
if err != nil {
@@ -248,19 +255,19 @@ func (st *ServerStream) WritePacketRTP(trackID int, pkt *rtp.Packet, ptsEqualsDT
}
track := st.streamTracks[trackID]
now := time.Now()
ptsEqualsDTS := ptsEqualsDTS(st.tracks[trackID], pkt)
if ptsEqualsDTS {
track.lastTimeFilled = true
track.lastTimeRTP = pkt.Header.Timestamp
track.lastTimeNTP = now
track.lastTimeNTP = ntp
}
track.lastSequenceNumber = pkt.Header.SequenceNumber
track.lastSSRC = pkt.Header.SSRC
if track.rtcpSender != nil {
track.rtcpSender.ProcessPacketRTP(now, pkt, ptsEqualsDTS)
track.rtcpSender.ProcessPacketRTP(ntp, pkt, ptsEqualsDTS)
}
// send unicast

1
serverudpl.go
View File

@@ -237,7 +237,6 @@ func (u *serverUDPListener) processRTP(clientData *clientData, payload []byte) {
Session: clientData.session,
TrackID: clientData.track.id,
Packet: pkt,
PTSEqualsDTS: ptsEqualsDTS,
})
}
}