github-forks
/
lal
mirror of https://github.com/q191201771/lal.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

messages:

Browse Source 
- [feat] rtsp，pull支持interleaved(over tcp)模式。demo/pullrtsp2pushrtmp和demo/pullrtsp可通过运行参数选择是否使用over tcp模式
- [feat] rtsp，pull支持auth digest验证
- [feat] rtsp，pull支持定时发送GET_PARAMETER rtsp message进行保活（对端支持的情况下）
- [fix] rtsp，输入流音频不是AAC格式时，保证视频流可正常remux成其他封装协议
- [fix] rtsp，pull开始时发送dummy rtp/rtcp数据，保证对端能成功发送数据至本地
- [fix] rtsp，将以下包返回给上层：rtsp pub h265, single rtp packet, VPS, SPS, PPS, SEI
- [fix] rtsp，修改rtsp.AVPacketQueue的行为：当音频或者视频数量队列满了后，直接出队而不是丢弃
- [fix] sdp，修复解析及使用sdp错误的一些case
- [refactor] rtmp/httpflv/rtsp，统一所有PullSession：超时形式；Pull和Wait函数
- [fix] avc，尝试解析scaling matrix
pull/44/head
q191201771 4 years ago
parent 3ca25a22d2
commit 942b5da52d
37 changed files with 1355 additions and 441 deletions
Show Stats Download Patch File Download Diff File

1
app/demo/analyseflv/analyseflv.go
Unescape Escape View File

@ -121,6 +121,7 @@ func main() {
nazalog.Debugf("%+v", buf.String()) nazalog.Debugf("%+v", buf.String())
} }
case httpflv.TagTypeAudio: case httpflv.TagTypeAudio:
nazalog.Debugf("header=%+v, %+v", tag.Header, tag.IsAACSeqHeader())
brAudio.Add(len(tag.Raw)) brAudio.Add(len(tag.Raw))
if tag.IsAACSeqHeader() { if tag.IsAACSeqHeader() {

5
app/demo/pullrtmp/pullrtmp.go
Unescape Escape View File

@ -73,8 +73,7 @@ func pull(url string, filename string) {
} }
session := rtmp.NewPullSession(func(option *rtmp.PullSessionOption) { session := rtmp.NewPullSession(func(option *rtmp.PullSessionOption) {
option.ConnectTimeoutMS = 3000 option.PullTimeoutMS = 30000
option.PullTimeoutMS = 3000
option.ReadAVTimeoutMS = 10000 option.ReadAVTimeoutMS = 10000
}) })
@ -89,7 +88,7 @@ func pull(url string, filename string) {
} }
}) })
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = <-session.Done() err = <-session.Wait()
nazalog.Debug(err) nazalog.Debug(err)
} }

2
app/demo/pullrtmp2hls/pullrtmp2hls.go
Unescape Escape View File

@ -48,7 +48,7 @@ func main() {
nazalog.Errorf("pull error. err=%+v", err) nazalog.Errorf("pull error. err=%+v", err)
os.Exit(-1) os.Exit(-1)
} }
err = <-pullSession.Done() err = <-pullSession.Wait()
nazalog.Errorf("pull error. err=%+v", err) nazalog.Errorf("pull error. err=%+v", err)
} }

48
app/demo/pullrtsp/pullrtsp.go
Unescape Escape View File

@ -12,6 +12,7 @@ import (
"flag" "flag"
"fmt" "fmt"
"os" "os"
"time"
"github.com/q191201771/lal/pkg/base" "github.com/q191201771/lal/pkg/base"
"github.com/q191201771/lal/pkg/httpflv" "github.com/q191201771/lal/pkg/httpflv"
@ -31,14 +32,18 @@ func (o *Observer) OnRTPPacket(pkt rtprtcp.RTPPacket) {
} }
func (o *Observer) OnAVConfig(asc, vps, sps, pps []byte) { func (o *Observer) OnAVConfig(asc, vps, sps, pps []byte) {
metadata, vsh, ash, err := remux.AVConfig2FLVTag(asc, vps, sps, pps) metadata, ash, vsh, err := remux.AVConfig2FLVTag(asc, vps, sps, pps)
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = w.WriteTag(*metadata) err = w.WriteTag(*metadata)
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = w.WriteTag(*vsh) if ash != nil {
nazalog.Assert(nil, err) err = w.WriteTag(*ash)
err = w.WriteTag(*ash) nazalog.Assert(nil, err)
nazalog.Assert(nil, err) }
if vsh != nil {
err = w.WriteTag(*vsh)
nazalog.Assert(nil, err)
}
} }
func (o *Observer) OnAVPacket(pkt base.AVPacket) { func (o *Observer) OnAVPacket(pkt base.AVPacket) {
@ -49,7 +54,11 @@ func (o *Observer) OnAVPacket(pkt base.AVPacket) {
} }
func main() { func main() {
inURL, outFilename := parseFlag() _ = nazalog.Init(func(option *nazalog.Option) {
option.AssertBehavior = nazalog.AssertFatal
})
inURL, outFilename, overTCP := parseFlag()
err := w.Open(outFilename) err := w.Open(outFilename)
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
defer w.Dispose() defer w.Dispose()
@ -57,23 +66,38 @@ func main() {
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
o := &Observer{} o := &Observer{}
s := rtsp.NewPullSession(o, func(option *rtsp.PullSessionOption) { rtspPullSession := rtsp.NewPullSession(o, func(option *rtsp.PullSessionOption) {
option.PullTimeoutMS = 5000 option.PullTimeoutMS = 5000
option.OverTCP = overTCP != 0
}) })
err = s.Pull(inURL)
nazalog.Error(err) go func() {
for {
rtspPullSession.UpdateStat(1)
rtspStat := rtspPullSession.GetStat()
nazalog.Debugf("bitrate. rtsp pull=%dkbit/s", rtspStat.Bitrate)
time.Sleep(1 * time.Second)
}
}()
err = rtspPullSession.Pull(inURL)
nazalog.Assert(nil, err)
err = <-rtspPullSession.Wait()
nazalog.Infof("done. err=%+v", err)
} }
func parseFlag() (inURL string, outFilename string) { func parseFlag() (inURL string, outFilename string, overTCP int) {
i := flag.String("i", "", "specify pull rtsp url") i := flag.String("i", "", "specify pull rtsp url")
o := flag.String("o", "", "specify ouput flv file") o := flag.String("o", "", "specify ouput flv file")
t := flag.Int("t", 0, "specify interleaved mode(rtp/rtcp over tcp)")
flag.Parse() flag.Parse()
if *i == "" || *o == "" { if *i == "" || *o == "" {
flag.Usage() flag.Usage()
_, _ = fmt.Fprintf(os.Stderr, `Example: _, _ = fmt.Fprintf(os.Stderr, `Example:
./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o out.flv ./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o out.flv -t 0
./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o out.flv -t 1
`) `)
os.Exit(1) os.Exit(1)
} }
return *i, *o return *i, *o, *t
} }

33
app/demo/pullrtsp2pushrtmp/pullrtsp2pushrtmp.go
Unescape Escape View File

@ -33,14 +33,18 @@ func (o *Observer) OnRTPPacket(pkt rtprtcp.RTPPacket) {
} }
func (o *Observer) OnAVConfig(asc, vps, sps, pps []byte) { func (o *Observer) OnAVConfig(asc, vps, sps, pps []byte) {
metadata, vsh, ash, err := remux.AVConfig2RTMPMsg(asc, vps, sps, pps) metadata, ash, vsh, err := remux.AVConfig2RTMPMsg(asc, vps, sps, pps)
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(metadata.Payload, &metadata.Header)) err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(metadata.Payload, &metadata.Header))
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(vsh.Payload, &vsh.Header)) if ash != nil {
nazalog.Assert(nil, err) err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(ash.Payload, &ash.Header))
err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(ash.Payload, &ash.Header)) nazalog.Assert(nil, err)
nazalog.Assert(nil, err) }
if vsh != nil {
err = rtmpPushSession.AsyncWrite(rtmp.Message2Chunks(vsh.Payload, &vsh.Header))
nazalog.Assert(nil, err)
}
} }
func (o *Observer) OnAVPacket(pkt base.AVPacket) { func (o *Observer) OnAVPacket(pkt base.AVPacket) {
@ -55,10 +59,9 @@ func main() {
option.AssertBehavior = nazalog.AssertFatal option.AssertBehavior = nazalog.AssertFatal
}) })
inURL, outURL := parseFlag() inURL, outURL, overTCP := parseFlag()
rtmpPushSession = rtmp.NewPushSession(func(option *rtmp.PushSessionOption) { rtmpPushSession = rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
option.ConnectTimeoutMS = 5000
option.PushTimeoutMS = 5000 option.PushTimeoutMS = 5000
option.WriteAVTimeoutMS = 5000 option.WriteAVTimeoutMS = 5000
}) })
@ -67,13 +70,15 @@ func main() {
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
go func() { go func() {
err := <-rtmpPushSession.Done() err := <-rtmpPushSession.Wait()
nazalog.Assert(nil, err) nazalog.Infof("push rtmp done. err=%+v", err)
os.Exit(1)
}() }()
o := &Observer{} o := &Observer{}
rtspPullSession := rtsp.NewPullSession(o, func(option *rtsp.PullSessionOption) { rtspPullSession := rtsp.NewPullSession(o, func(option *rtsp.PullSessionOption) {
option.PullTimeoutMS = 5000 option.PullTimeoutMS = 5000
option.OverTCP = overTCP != 0
}) })
go func() { go func() {
@ -89,18 +94,22 @@ func main() {
err = rtspPullSession.Pull(inURL) err = rtspPullSession.Pull(inURL)
nazalog.Assert(nil, err) nazalog.Assert(nil, err)
err = <-rtspPullSession.Wait()
nazalog.Infof("pull rtsp done. err=%+v", err)
} }
func parseFlag() (inURL string, outFilename string) { func parseFlag() (inURL string, outFilename string, overTCP int) {
i := flag.String("i", "", "specify pull rtsp url") i := flag.String("i", "", "specify pull rtsp url")
o := flag.String("o", "", "specify push rtmp url") o := flag.String("o", "", "specify push rtmp url")
t := flag.Int("t", 0, "specify interleaved mode(rtp/rtcp over tcp)")
flag.Parse() flag.Parse()
if *i == "" || *o == "" { if *i == "" || *o == "" {
flag.Usage() flag.Usage()
_, _ = fmt.Fprintf(os.Stderr, `Example: _, _ = fmt.Fprintf(os.Stderr, `Example:
./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o rtmp://localhost:19350/live/test220 ./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o rtmp://localhost:19350/live/test220 -t 0
./bin/pullrtsp -i rtsp://localhost:5544/live/test110 -o rtmp://localhost:19350/live/test220 -t 1
`) `)
os.Exit(1) os.Exit(1)
} }
return *i, *o return *i, *o, *t
} }

1
app/demo/pushrtmp/pushrtmp.go
Unescape Escape View File

@ -131,7 +131,6 @@ func push(tags []httpflv.Tag, urls []string, isRecursive bool) {
for i := range urls { for i := range urls {
ps := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) { ps := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
option.ConnectTimeoutMS = 3000
option.PushTimeoutMS = 5000 option.PushTimeoutMS = 5000
option.WriteAVTimeoutMS = 10000 option.WriteAVTimeoutMS = 10000
}) })

6
pkg/aac/aac.go
Unescape Escape View File

@ -62,9 +62,9 @@ func (a *ADTS) InitWithAACAudioSpecificConfig(asc []byte) error {
// <1.6.3.3 samplingFrequencyIndex>, <page 35/110> // <1.6.3.3 samplingFrequencyIndex>, <page 35/110>
// <1.6.3.4 channelConfiguration> // <1.6.3.4 channelConfiguration>
// -------------------------------------------------------- // --------------------------------------------------------
// audio object type [5b] 2=AAC LC // audio object type [5b] 1=AAC MAIN 2=AAC LC
// samplingFrequencyIndex [4b] 3=48000 4=44100 // samplingFrequencyIndex [4b] 3=48000 4=44100 6=24000 5=32000 11=11025
// channelConfiguration [4b] 2=left, right front speakers // channelConfiguration [4b] 1=center front speaker 2=left, right front speakers
br := nazabits.NewBitReader(asc) br := nazabits.NewBitReader(asc)
a.audioObjectType, _ = br.ReadBits8(5) a.audioObjectType, _ = br.ReadBits8(5)
a.samplingFrequencyIndex, _ = br.ReadBits8(4) a.samplingFrequencyIndex, _ = br.ReadBits8(4)

143
pkg/avc/avc.go
Unescape Escape View File

@ -425,8 +425,12 @@ func ParseSPS(payload []byte) (Context, error) {
return Context{}, err return Context{}, err
} }
if flag == 1 { if flag == 1 {
nazalog.Debugf("scaling matrix present, not impl yet.") nazalog.Debugf("scaling matrix present.")
return Context{}, ErrAVC // TODO chef: 还没有正确实现，只是针对特定case做了处理
_, err = br.ReadBits32(128)
if err != nil {
return Context{}, err
}
} }
} else { } else {
sps.ChromaFormatIdc = 1 sps.ChromaFormatIdc = 1
@ -438,10 +442,14 @@ func ParseSPS(payload []byte) (Context, error) {
if err != nil { if err != nil {
return Context{}, err return Context{}, err
} }
if sps.Log2MaxFrameNumMinus4 > 12 {
return Context{}, ErrAVC
}
sps.PicOrderCntType, err = br.ReadGolomb() sps.PicOrderCntType, err = br.ReadGolomb()
if err != nil { if err != nil {
return Context{}, err return Context{}, err
} }
if sps.PicOrderCntType == 0 { if sps.PicOrderCntType == 0 {
sps.Log2MaxPicOrderCntLsb, err = br.ReadGolomb() sps.Log2MaxPicOrderCntLsb, err = br.ReadGolomb()
sps.Log2MaxPicOrderCntLsb += 4 sps.Log2MaxPicOrderCntLsb += 4
@ -579,3 +587,134 @@ func TryParseSeqHeader(payload []byte) error {
return err return err
} }
//var defaultScaling4 = [][]uint8{
// {
// 6, 13, 20, 28, 13, 20, 28, 32,
// 20, 28, 32, 37, 28, 32, 37, 42,
// },
// {
// 10, 14, 20, 24, 14, 20, 24, 27,
// 20, 24, 27, 30, 24, 27, 30, 34,
// },
//}
//
//var defaultScaling8 = [][]uint8{
// {
// 6, 10, 13, 16, 18, 23, 25, 27,
// 10, 11, 16, 18, 23, 25, 27, 29,
// 13, 16, 18, 23, 25, 27, 29, 31,
// 16, 18, 23, 25, 27, 29, 31, 33,
// 18, 23, 25, 27, 29, 31, 33, 36,
// 23, 25, 27, 29, 31, 33, 36, 38,
// 25, 27, 29, 31, 33, 36, 38, 40,
// 27, 29, 31, 33, 36, 38, 40, 42,
// },
// {
// 9, 13, 15, 17, 19, 21, 22, 24,
// 13, 13, 17, 19, 21, 22, 24, 25,
// 15, 17, 19, 21, 22, 24, 25, 27,
// 17, 19, 21, 22, 24, 25, 27, 28,
// 19, 21, 22, 24, 25, 27, 28, 30,
// 21, 22, 24, 25, 27, 28, 30, 32,
// 22, 24, 25, 27, 28, 30, 32, 33,
// 24, 25, 27, 28, 30, 32, 33, 35,
// },
//}
//
//var ffZigzagDirect = []uint8{
// 0, 1, 8, 16, 9, 2, 3, 10,
// 17, 24, 32, 25, 18, 11, 4, 5,
// 12, 19, 26, 33, 40, 48, 41, 34,
// 27, 20, 13, 6, 7, 14, 21, 28,
// 35, 42, 49, 56, 57, 50, 43, 36,
// 29, 22, 15, 23, 30, 37, 44, 51,
// 58, 59, 52, 45, 38, 31, 39, 46,
// 53, 60, 61, 54, 47, 55, 62, 63,
//}
//
//var ffZigzagScan = []uint8{
// 0 + 0*4, 1 + 0*4, 0 + 1*4, 0 + 2*4,
// 1 + 1*4, 2 + 0*4, 3 + 0*4, 2 + 1*4,
// 1 + 2*4, 0 + 3*4, 1 + 3*4, 2 + 2*4,
// 3 + 1*4, 3 + 2*4, 2 + 3*4, 3 + 3*4,
//}
//
//func decodeScalingMatrices(reader *nazabits.BitReader) error {
// // 6 * 16
// var spsScalingMatrix4 = [][]uint8{
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// }
// // 6 * 64
// var spsScalingMatrix8 = [][]uint8{
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0},
// }
//
// fallback := [][]uint8{defaultScaling4[0], defaultScaling4[1], defaultScaling8[0], defaultScaling8[1]}
// decodeScalingList(reader, spsScalingMatrix4[0], 16, defaultScaling4[0], fallback[0])
// decodeScalingList(reader, spsScalingMatrix4[1], 16, defaultScaling4[0], spsScalingMatrix4[0])
// decodeScalingList(reader, spsScalingMatrix4[2], 16, defaultScaling4[0], spsScalingMatrix4[1])
// decodeScalingList(reader, spsScalingMatrix4[3], 16, defaultScaling4[1], fallback[1])
// decodeScalingList(reader, spsScalingMatrix4[4], 16, defaultScaling4[1], spsScalingMatrix4[3])
// decodeScalingList(reader, spsScalingMatrix4[4], 16, defaultScaling4[1], spsScalingMatrix4[3])
//
// decodeScalingList(reader, spsScalingMatrix8[0], 64, defaultScaling8[0], fallback[2])
// decodeScalingList(reader, spsScalingMatrix8[3], 64, defaultScaling8[1], fallback[3])
//
// return nil
//}
//
//func decodeScalingList(reader *nazabits.BitReader, factors []uint8, size int, jvtList []uint8, fallbackList []uint8) error {
// var (
// i = 0
// last = 8
// next = 8
// scan []uint8
// )
// if size == 16 {
// scan = ffZigzagScan
// } else {
// scan = ffZigzagDirect
// }
// flag, err := reader.ReadBit()
// if err != nil {
// return err
// }
// return nil
// if flag == 0 {
// for n := 0; n < size; n++ {
// factors[n] = fallbackList[n]
// }
// } else {
// for i = 0; i < size; i++ {
// if next != 0 {
// v, err := reader.ReadGolomb()
// if err != nil {
// return err
// }
// next = (last + int(v)) & 0xff
// }
// if i == 0 && next == 0 {
// for n := 0; n < size; n++ {
// factors[n] = jvtList[n]
// }
// break
// }
// if next != 0 {
// factors[scan[i]] = uint8(next)
// last = next
// } else {
// factors[scan[i]] = uint8(last)
// }
// }
// }
// return nil
//}

10
pkg/avc/avc_test.go
Unescape Escape View File

@ -134,3 +134,13 @@ func TestCorner(t *testing.T) {
assert.Equal(t, nil, b.Bytes()) assert.Equal(t, nil, b.Bytes())
assert.Equal(t, avc.ErrAVC, err) assert.Equal(t, avc.ErrAVC, err)
} }
func TestParsePPS_Case2(t *testing.T) {
in := []byte{0x67, 0x64, 0x00, 0x20, 0xad, 0x84, 0x01, 0x0c, 0x20, 0x08, 0x61, 0x00, 0x43, 0x08, 0x02, 0x18, 0x40, 0x10, 0xc2, 0x00, 0x84, 0x3b, 0x50, 0x28, 0x03, 0xcd, 0x37, 0x01, 0x01, 0x01, 0x40, 0x00, 0x00, 0x03, 0x00, 0x40, 0x00, 0x00, 0x0c, 0xa1}
ctx, err := avc.ParseSPS(in)
assert.Equal(t, nil, err)
assert.Equal(t, uint8(100), ctx.Profile)
assert.Equal(t, uint8(32), ctx.Level)
assert.Equal(t, uint32(1280), ctx.Width)
assert.Equal(t, uint32(960), ctx.Height)
}

7
pkg/base/avpacket.go
Unescape Escape View File

@ -11,9 +11,10 @@ package base
type AVPacketPT int type AVPacketPT int
const ( const (
AVPacketPTAVC AVPacketPT = RTPPacketTypeAVCOrHEVC AVPacketPTUnknown AVPacketPT = -1
AVPacketPTAAC AVPacketPT = RTPPacketTypeAAC AVPacketPTAVC AVPacketPT = RTPPacketTypeAVCOrHEVC
AVPacketPTHEVC AVPacketPT = 98 AVPacketPTHEVC AVPacketPT = RTPPacketTypeHEVC
AVPacketPTAAC AVPacketPT = RTPPacketTypeAAC
) )
// 目前供package rtsp使用。以后可能被多个package使用。 // 目前供package rtsp使用。以后可能被多个package使用。

6
pkg/base/rtprtcp.go
Unescape Escape View File

@ -9,7 +9,11 @@
package base package base
const ( const (
// 注意，AVC和HEVC都可能使用96，所以不能直接通过96判断是AVC还是HEVC // 注意，一般情况下，AVC使用96，AAC使用97，HEVC使用98
// 但是我还遇到过：
// HEVC使用96
// AVC使用105
RTPPacketTypeAVCOrHEVC = 96 RTPPacketTypeAVCOrHEVC = 96
RTPPacketTypeAAC = 97 RTPPacketTypeAAC = 97
RTPPacketTypeHEVC = 98
) )

5
pkg/base/session.go
Unescape Escape View File

@ -74,12 +74,15 @@ type ISessionURLContext interface {
// | UpdateStat()<all> | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | | // | UpdateStat()<all> | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | |
// | IsAlive()<all> | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | | // | IsAlive()<all> | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | |
// | RunLoop() | √ | x√ | √ | √ | √ | x√ | - | x | x | x | x | | // | RunLoop() | √ | x√ | √ | √ | √ | x&√ | - | x | x | x | x | |
// | Dispose() | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | | // | Dispose() | √ | √ | √ | √ | √ | √ | - | √ | √ | √ | √ | |
// | RemoteAddr() | √ | x | √ | √ | x | x | - | x | x | x | x | | // | RemoteAddr() | √ | x | √ | √ | x | x | - | x | x | x | x | |
// | SingleConn | √ | x | √ | √ | √ | √ | - | √ | √ | √ | x | | // | SingleConn | √ | x | √ | √ | √ | √ | - | √ | √ | √ | x | |
// //
// | Opt.PullTimeoutMS | - | - | - | - | - | - | - | - | x | √ | √ | |
// | Wait() | - | - | - | - | - | - | - | - | √ | √ | √ | |
//
// Dispose由外部调用，表示主动关闭正常的session // Dispose由外部调用，表示主动关闭正常的session
// 外部调用Dispose后，不应继续使用该session // 外部调用Dispose后，不应继续使用该session
// Dispose后，RunLoop结束阻塞 // Dispose后，RunLoop结束阻塞

10
pkg/base/url.go
Unescape Escape View File

@ -19,6 +19,8 @@ import (
// 见单元测试 // 见单元测试
// TODO chef: 考虑部分内容移入naza中
var ErrURL = errors.New("lal.url: fxxk") var ErrURL = errors.New("lal.url: fxxk")
const ( const (
@ -37,6 +39,8 @@ type URLPathContext struct {
type URLContext struct { type URLContext struct {
Scheme string Scheme string
Username string
Password string
StdHost string // host or host:port StdHost string // host or host:port
HostWithPort string HostWithPort string
Host string Host string
@ -48,6 +52,8 @@ type URLContext struct {
PathWithoutLastItem string // 注意，没有前面的'/'，也没有后面的'/' PathWithoutLastItem string // 注意，没有前面的'/'，也没有后面的'/'
LastItemOfPath string // 注意，没有前面的'/' LastItemOfPath string // 注意，没有前面的'/'
RawQuery string RawQuery string
RawURLWithoutUserInfo string
} }
func ParseURLPath(path string) (ctx URLPathContext, err error) { func ParseURLPath(path string) (ctx URLPathContext, err error) {
@ -69,6 +75,8 @@ func ParseURL(rawURL string, defaultPort int) (ctx URLContext, err error) {
ctx.Scheme = stdURL.Scheme ctx.Scheme = stdURL.Scheme
ctx.StdHost = stdURL.Host ctx.StdHost = stdURL.Host
ctx.Username = stdURL.User.Username()
ctx.Password, _ = stdURL.User.Password()
h, p, err := net.SplitHostPort(stdURL.Host) h, p, err := net.SplitHostPort(stdURL.Host)
if err != nil { if err != nil {
@ -102,6 +110,8 @@ func ParseURL(rawURL string, defaultPort int) (ctx URLContext, err error) {
ctx.PathWithoutLastItem = pathCtx.PathWithoutLastItem ctx.PathWithoutLastItem = pathCtx.PathWithoutLastItem
ctx.LastItemOfPath = pathCtx.LastItemOfPath ctx.LastItemOfPath = pathCtx.LastItemOfPath
ctx.RawQuery = pathCtx.RawQuery ctx.RawQuery = pathCtx.RawQuery
ctx.RawURLWithoutUserInfo = fmt.Sprintf("%s://%s%s", ctx.Scheme, ctx.StdHost, ctx.PathWithRawQuery)
return ctx, nil return ctx, nil
} }

257
pkg/base/url_test.go
Unescape Escape View File

@ -30,133 +30,143 @@ func TestParseURL(t *testing.T) {
golden := map[in]base.URLContext{ golden := map[in]base.URLContext{
// 常见url，url中无端口，另外设置默认端口 // 常见url，url中无端口，另外设置默认端口
in{rawURL: "rtmp://127.0.0.1/live/test110", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1/live/test110", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1", StdHost: "127.0.0.1",
HostWithPort: "127.0.0.1:1935", HostWithPort: "127.0.0.1:1935",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 1935, Port: 1935,
PathWithRawQuery: "/live/test110", PathWithRawQuery: "/live/test110",
Path: "/live/test110", Path: "/live/test110",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1/live/test110",
}, },
// 域名url // 域名url
in{rawURL: "rtmp://localhost/live/test110", defaultPort: 1935}: { in{rawURL: "rtmp://localhost/live/test110", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "localhost", StdHost: "localhost",
HostWithPort: "localhost:1935", HostWithPort: "localhost:1935",
Host: "localhost", Host: "localhost",
Port: 1935, Port: 1935,
PathWithRawQuery: "/live/test110", PathWithRawQuery: "/live/test110",
Path: "/live/test110", Path: "/live/test110",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://localhost/live/test110",
}, },
// 带参数url // 带参数url
in{rawURL: "rtmp://127.0.0.1/live/test110?a=1", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1/live/test110?a=1", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1", StdHost: "127.0.0.1",
HostWithPort: "127.0.0.1:1935", HostWithPort: "127.0.0.1:1935",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 1935, Port: 1935,
PathWithRawQuery: "/live/test110?a=1", PathWithRawQuery: "/live/test110?a=1",
Path: "/live/test110", Path: "/live/test110",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "a=1", RawQuery: "a=1",
RawURLWithoutUserInfo: "rtmp://127.0.0.1/live/test110?a=1",
}, },
// path多级 // path多级
in{rawURL: "rtmp://127.0.0.1:19350/a/b/test110", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350/a/b/test110", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "/a/b/test110", PathWithRawQuery: "/a/b/test110",
Path: "/a/b/test110", Path: "/a/b/test110",
PathWithoutLastItem: "a/b", PathWithoutLastItem: "a/b",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350/a/b/test110",
}, },
// url中无端口，没有设置默认端口 // url中无端口，没有设置默认端口
in{rawURL: "rtmp://127.0.0.1/live/test110?a=1", defaultPort: -1}: { in{rawURL: "rtmp://127.0.0.1/live/test110?a=1", defaultPort: -1}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1", StdHost: "127.0.0.1",
HostWithPort: "127.0.0.1", HostWithPort: "127.0.0.1",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 0, Port: 0,
PathWithRawQuery: "/live/test110?a=1", PathWithRawQuery: "/live/test110?a=1",
Path: "/live/test110", Path: "/live/test110",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "a=1", RawQuery: "a=1",
RawURLWithoutUserInfo: "rtmp://127.0.0.1/live/test110?a=1",
}, },
// url中有端口，设置默认端口 // url中有端口，设置默认端口
in{rawURL: "rtmp://127.0.0.1:19350/live/test110?a=1", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350/live/test110?a=1", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "/live/test110?a=1", PathWithRawQuery: "/live/test110?a=1",
Path: "/live/test110", Path: "/live/test110",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "test110", LastItemOfPath: "test110",
RawQuery: "a=1", RawQuery: "a=1",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350/live/test110?a=1",
}, },
// 无path // 无path
in{rawURL: "rtmp://127.0.0.1:19350", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "", PathWithRawQuery: "",
Path: "", Path: "",
PathWithoutLastItem: "", PathWithoutLastItem: "",
LastItemOfPath: "", LastItemOfPath: "",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350",
}, },
// 无path2 // 无path2
in{rawURL: "rtmp://127.0.0.1:19350/", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350/", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "/", PathWithRawQuery: "/",
Path: "/", Path: "/",
PathWithoutLastItem: "", PathWithoutLastItem: "",
LastItemOfPath: "", LastItemOfPath: "",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350/",
}, },
// path不完整 // path不完整
in{rawURL: "rtmp://127.0.0.1:19350/live", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350/live", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "/live", PathWithRawQuery: "/live",
Path: "/live", Path: "/live",
PathWithoutLastItem: "", PathWithoutLastItem: "",
LastItemOfPath: "live", LastItemOfPath: "live",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350/live",
}, },
// path不完整2 // path不完整2
in{rawURL: "rtmp://127.0.0.1:19350/live/", defaultPort: 1935}: { in{rawURL: "rtmp://127.0.0.1:19350/live/", defaultPort: 1935}: {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1:19350", StdHost: "127.0.0.1:19350",
HostWithPort: "127.0.0.1:19350", HostWithPort: "127.0.0.1:19350",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 19350, Port: 19350,
PathWithRawQuery: "/live/", PathWithRawQuery: "/live/",
Path: "/live/", Path: "/live/",
PathWithoutLastItem: "live", PathWithoutLastItem: "live",
LastItemOfPath: "", LastItemOfPath: "",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1:19350/live/",
}, },
} }
@ -171,16 +181,17 @@ func TestParseRTMPURL(t *testing.T) {
golden := map[string]base.URLContext{ golden := map[string]base.URLContext{
// 其他测试见ParseURL // 其他测试见ParseURL
"rtmp://127.0.0.1/test110": { "rtmp://127.0.0.1/test110": {
Scheme: "rtmp", Scheme: "rtmp",
StdHost: "127.0.0.1", StdHost: "127.0.0.1",
HostWithPort: "127.0.0.1:1935", HostWithPort: "127.0.0.1:1935",
Host: "127.0.0.1", Host: "127.0.0.1",
Port: 1935, Port: 1935,
PathWithRawQuery: "/test110", PathWithRawQuery: "/test110",
Path: "/test110", Path: "/test110",
PathWithoutLastItem: "test110", PathWithoutLastItem: "test110",
LastItemOfPath: "", LastItemOfPath: "",
RawQuery: "", RawQuery: "",
RawURLWithoutUserInfo: "rtmp://127.0.0.1/test110",
}, },
} }
for k, v := range golden { for k, v := range golden {
@ -189,3 +200,29 @@ func TestParseRTMPURL(t *testing.T) {
assert.Equal(t, v, ctx, k) assert.Equal(t, v, ctx, k)
} }
} }
func TestParseRTSPURL(t *testing.T) {
golden := map[string]base.URLContext{
// 其他测试见ParseURL
"rtsp://admin:P!@1988@127.0.0.1:5554/h264/ch33/main/av_stream": {
Scheme: "rtsp",
Username: "admin",
Password: "P!@1988",
StdHost: "127.0.0.1:5554",
HostWithPort: "127.0.0.1:5554",
Host: "127.0.0.1",
Port: 5554,
PathWithRawQuery: "/h264/ch33/main/av_stream",
Path: "/h264/ch33/main/av_stream",
PathWithoutLastItem: "h264/ch33/main",
LastItemOfPath: "av_stream",
RawQuery: "",
RawURLWithoutUserInfo: "rtsp://127.0.0.1:5554/h264/ch33/main/av_stream",
},
}
for k, v := range golden {
ctx, err := base.ParseRTSPURL(k)
assert.Equal(t, nil, err)
assert.Equal(t, v, ctx, k)
}
}

85
pkg/httpflv/client_pull_session.go
Unescape Escape View File

@ -9,6 +9,7 @@
package httpflv package httpflv
import ( import (
"context"
"fmt" "fmt"
"net" "net"
"time" "time"
@ -22,13 +23,16 @@ import (
) )
type PullSessionOption struct { type PullSessionOption struct {
ConnectTimeoutMS int // TCP连接时超时，单位毫秒，如果为0，则不设置超时 // 从调用Pull函数，到接收音视频数据的前一步，也即发送完HTTP请求的超时时间
ReadTimeoutMS int // 接收数据超时，单位毫秒，如果为0，则不设置超时 // 如果为0，则没有超时时间
PullTimeoutMS int
ReadTimeoutMS int // 接收数据超时，单位毫秒，如果为0，则不设置超时
} }
var defaultPullSessionOption = PullSessionOption{ var defaultPullSessionOption = PullSessionOption{
ConnectTimeoutMS: 0, PullTimeoutMS: 10000,
ReadTimeoutMS: 0, ReadTimeoutMS: 0,
} }
type PullSession struct { type PullSession struct {
@ -41,6 +45,8 @@ type PullSession struct {
stat base.StatSession stat base.StatSession
urlCtx base.URLContext urlCtx base.URLContext
waitErrChan chan error
} }
type ModPullSessionOption func(option *PullSessionOption) type ModPullSessionOption func(option *PullSessionOption)
@ -53,8 +59,9 @@ func NewPullSession(modOptions ...ModPullSessionOption) *PullSession {
uk := base.GenUniqueKey(base.UKPFLVPullSession) uk := base.GenUniqueKey(base.UKPFLVPullSession)
s := &PullSession{ s := &PullSession{
option: option, UniqueKey: uk,
UniqueKey: uk, option: option,
waitErrChan: make(chan error, 1),
} }
nazalog.Infof("[%s] lifecycle new httpflv PullSession. session=%p", uk, s) nazalog.Infof("[%s] lifecycle new httpflv PullSession. session=%p", uk, s)
return s return s
@ -62,22 +69,59 @@ func NewPullSession(modOptions ...ModPullSessionOption) *PullSession {
type OnReadFLVTag func(tag Tag) type OnReadFLVTag func(tag Tag)
// 阻塞直到拉流失败 // 如果没有错误发生，阻塞直到接收音视频数据的前一步，也即发送完HTTP请求
// //
// @param rawURL 支持如下两种格式。（当然，关键点是对端支持） // @param rawURL 支持如下两种格式。（当然，关键点是对端支持）
// http://{domain}/{app_name}/{stream_name}.flv // http://{domain}/{app_name}/{stream_name}.flv
// http://{ip}/{domain}/{app_name}/{stream_name}.flv // http://{ip}/{domain}/{app_name}/{stream_name}.flv
// //
// @param onReadFLVTag 读取到 flv tag 数据时回调。回调结束后，PullSession 不会再使用这块 <tag> 数据。 // @param onReadFLVTag 读取到 flv tag 数据时回调。回调结束后，PullSession 不会再使用这块 <tag> 数据。
func (session *PullSession) Pull(rawURL string, onReadFLVTag OnReadFLVTag) error { func (session *PullSession) Pull(rawURL string, onReadFLVTag OnReadFLVTag) error {
if err := session.connect(rawURL); err != nil { var (
return err ctx context.Context
cancel context.CancelFunc
)
if session.option.PullTimeoutMS == 0 {
ctx, cancel = context.WithCancel(context.Background())
} else {
ctx, cancel = context.WithTimeout(context.Background(), time.Duration(session.option.PullTimeoutMS)*time.Millisecond)
} }
if err := session.writeHTTPRequest(); err != nil { defer cancel()
return err return session.pullContext(ctx, rawURL, onReadFLVTag)
}
// Pull成功后，调用该函数，可阻塞直到拉流结束
func (session *PullSession) Wait() <-chan error {
return session.waitErrChan
}
func (session *PullSession) pullContext(ctx context.Context, rawURL string, onReadFLVTag OnReadFLVTag) error {
errChan := make(chan error, 1)
go func() {
if err := session.connect(rawURL); err != nil {
errChan <- err
return
}
if err := session.writeHTTPRequest(); err != nil {
errChan <- err
return
}
errChan <- nil
}()
select {
case <-ctx.Done():
return ctx.Err()
case err := <-errChan:
if err != nil {
return err
}
} }
return session.runReadLoop(onReadFLVTag) go session.runReadLoop(onReadFLVTag)
return nil
} }
func (session *PullSession) Dispose() { func (session *PullSession) Dispose() {
@ -137,7 +181,7 @@ func (session *PullSession) connect(rawURL string) (err error) {
nazalog.Debugf("[%s] > tcp connect.", session.UniqueKey) nazalog.Debugf("[%s] > tcp connect.", session.UniqueKey)
// # 建立连接 // # 建立连接
conn, err := net.DialTimeout("tcp", session.urlCtx.HostWithPort, time.Duration(session.option.ConnectTimeoutMS)*time.Millisecond) conn, err := net.Dial("tcp", session.urlCtx.HostWithPort)
if err != nil { if err != nil {
return err return err
} }
@ -188,19 +232,22 @@ func (session *PullSession) readTag() (Tag, error) {
return readTag(session.conn) return readTag(session.conn)
} }
func (session *PullSession) runReadLoop(onReadFLVTag OnReadFLVTag) error { func (session *PullSession) runReadLoop(onReadFLVTag OnReadFLVTag) {
if _, _, err := session.readHTTPRespHeader(); err != nil { if _, _, err := session.readHTTPRespHeader(); err != nil {
return err session.waitErrChan <- err
return
} }
if _, err := session.readFLVHeader(); err != nil { if _, err := session.readFLVHeader(); err != nil {
return err session.waitErrChan <- err
return
} }
for { for {
tag, err := session.readTag() tag, err := session.readTag()
if err != nil { if err != nil {
return err session.waitErrChan <- err
return
} }
onReadFLVTag(tag) onReadFLVTag(tag)
} }

2
pkg/innertest/innertest.go
Unescape Escape View File

@ -114,7 +114,7 @@ func InnerTestEntry(t *testing.T) {
if err != nil { if err != nil {
nazalog.Error(err) nazalog.Error(err)
} }
err = <-rtmpPullSession.Done() err = <-rtmpPullSession.Wait()
nazalog.Debug(err) nazalog.Debug(err)
}() }()

6
pkg/logic/group.go
Unescape Escape View File

@ -848,7 +848,6 @@ func (group *Group) pullIfNeeded() {
go func() { go func() {
pullSession := rtmp.NewPullSession(func(option *rtmp.PullSessionOption) { pullSession := rtmp.NewPullSession(func(option *rtmp.PullSessionOption) {
option.ConnectTimeoutMS = relayPullConnectTimeoutMS
option.PullTimeoutMS = relayPullTimeoutMS option.PullTimeoutMS = relayPullTimeoutMS
option.ReadAVTimeoutMS = relayPullReadAVTimeoutMS option.ReadAVTimeoutMS = relayPullReadAVTimeoutMS
}) })
@ -860,7 +859,7 @@ func (group *Group) pullIfNeeded() {
} }
res := group.AddRTMPPullSession(pullSession) res := group.AddRTMPPullSession(pullSession)
if res { if res {
err = <-pullSession.Done() err = <-pullSession.Wait()
nazalog.Infof("[%s] relay pull done. err=%v", pullSession.UniqueKey(), err) nazalog.Infof("[%s] relay pull done. err=%v", pullSession.UniqueKey(), err)
group.DelRTMPPullSession(pullSession) group.DelRTMPPullSession(pullSession)
} else { } else {
@ -901,7 +900,6 @@ func (group *Group) pushIfNeeded() {
go func(u, u2 string) { go func(u, u2 string) {
pushSession := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) { pushSession := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
option.ConnectTimeoutMS = relayPushConnectTimeoutMS
option.PushTimeoutMS = relayPushTimeoutMS option.PushTimeoutMS = relayPushTimeoutMS
option.WriteAVTimeoutMS = relayPushWriteAVTimeoutMS option.WriteAVTimeoutMS = relayPushWriteAVTimeoutMS
}) })
@ -912,7 +910,7 @@ func (group *Group) pushIfNeeded() {
return return
} }
group.AddRTMPPushSession(u, pushSession) group.AddRTMPPushSession(u, pushSession)
err = <-pushSession.Done() err = <-pushSession.Wait()
nazalog.Infof("[%s] relay push done. err=%v", pushSession.UniqueKey(), err) nazalog.Infof("[%s] relay push done. err=%v", pushSession.UniqueKey(), err)
group.DelRTMPPushSession(u, pushSession) group.DelRTMPPushSession(u, pushSession)
}(url, urlWithParam) }(url, urlWithParam)

2
pkg/logic/var.go
Unescape Escape View File

@ -9,11 +9,9 @@
package logic package logic
//var relayPushCheckIntervalMS = 1000 //var relayPushCheckIntervalMS = 1000
var relayPushConnectTimeoutMS = 5000
var relayPushTimeoutMS = 5000 var relayPushTimeoutMS = 5000
var relayPushWriteAVTimeoutMS = 5000 var relayPushWriteAVTimeoutMS = 5000
var relayPullConnectTimeoutMS = 5000
var relayPullTimeoutMS = 5000 var relayPullTimeoutMS = 5000
var relayPullReadAVTimeoutMS = 5000 var relayPullReadAVTimeoutMS = 5000

22
pkg/remux/avpacket2flv.go
Unescape Escape View File

@ -16,6 +16,7 @@ import (
"github.com/q191201771/lal/pkg/httpflv" "github.com/q191201771/lal/pkg/httpflv"
"github.com/q191201771/lal/pkg/rtmp" "github.com/q191201771/lal/pkg/rtmp"
"github.com/q191201771/naza/pkg/bele" "github.com/q191201771/naza/pkg/bele"
"github.com/q191201771/naza/pkg/nazalog"
) )
// @param asc 如果为nil，则没有音频 // @param asc 如果为nil，则没有音频
@ -46,11 +47,14 @@ func AVConfig2FLVTag(asc, vps, sps, pps []byte) (metadata, ash, vsh *httpflv.Tag
if isHEVC { if isHEVC {
videocodecid = int(base.RTMPCodecIDHEVC) videocodecid = int(base.RTMPCodecIDHEVC)
var ctx hevc.Context var ctx hevc.Context
if err = hevc.ParseSPS(sps, &ctx); err != nil { err = hevc.ParseSPS(sps, &ctx)
return if err == nil {
width = int(ctx.PicWidthInLumaSamples)
height = int(ctx.PicHeightInLumaSamples)
} else {
// TODO chef: 如果解析错误，先忽略，将失败的case收集起来解决
nazalog.Warnf("parse sps failed.")
} }
width = int(ctx.PicWidthInLumaSamples)
height = int(ctx.PicHeightInLumaSamples)
bVsh, err = hevc.BuildSeqHeaderFromVPSSPSPPS(vps, sps, pps) bVsh, err = hevc.BuildSeqHeaderFromVPSSPSPPS(vps, sps, pps)
if err != nil { if err != nil {
return return
@ -59,11 +63,13 @@ func AVConfig2FLVTag(asc, vps, sps, pps []byte) (metadata, ash, vsh *httpflv.Tag
videocodecid = int(base.RTMPCodecIDAVC) videocodecid = int(base.RTMPCodecIDAVC)
var ctx avc.Context var ctx avc.Context
ctx, err = avc.ParseSPS(sps) ctx, err = avc.ParseSPS(sps)
if err != nil { if err == nil {
return width = int(ctx.Width)
height = int(ctx.Height)
} else {
// TODO chef: 如果解析错误，先忽略，将失败的case收集起来解决
nazalog.Warnf("parse sps failed.")
} }
width = int(ctx.Width)
height = int(ctx.Height)
bVsh, err = avc.BuildSeqHeaderFromSPSPPS(sps, pps) bVsh, err = avc.BuildSeqHeaderFromSPSPPS(sps, pps)
if err != nil { if err != nil {
return return

24
pkg/rtmp/client_pull_session.go
Unescape Escape View File

@ -19,15 +19,16 @@ type PullSession struct {
} }
type PullSessionOption struct { type PullSessionOption struct {
ConnectTimeoutMS int // 从调用Pull函数，到接收音视频数据的前一步，也即收到服务端返回的rtmp play对应结果的信令的超时时间
PullTimeoutMS int // 如果为0，则没有超时时间
ReadAVTimeoutMS int PullTimeoutMS int
ReadAVTimeoutMS int
} }
var defaultPullSessionOption = PullSessionOption{ var defaultPullSessionOption = PullSessionOption{
ConnectTimeoutMS: 0, PullTimeoutMS: 0,
PullTimeoutMS: 0, ReadAVTimeoutMS: 0,
ReadAVTimeoutMS: 0,
} }
type ModPullSessionOption func(option *PullSessionOption) type ModPullSessionOption func(option *PullSessionOption)
@ -40,24 +41,23 @@ func NewPullSession(modOptions ...ModPullSessionOption) *PullSession {
return &PullSession{ return &PullSession{
core: NewClientSession(CSTPullSession, func(option *ClientSessionOption) { core: NewClientSession(CSTPullSession, func(option *ClientSessionOption) {
option.ConnectTimeoutMS = opt.ConnectTimeoutMS
option.DoTimeoutMS = opt.PullTimeoutMS option.DoTimeoutMS = opt.PullTimeoutMS
option.ReadAVTimeoutMS = opt.ReadAVTimeoutMS option.ReadAVTimeoutMS = opt.ReadAVTimeoutMS
}), }),
} }
} }
// 建立rtmp play连接 // 如果没有发生错误，阻塞直到到接收音视频数据的前一步，也即收到服务端返回的rtmp play对应结果的信令
// 阻塞直到收到服务端返回的rtmp publish对应结果的信令，或发生错误
// //
// @param onReadRTMPAVMsg: 注意，回调结束后，内存块会被PullSession重复使用 // @param onReadRTMPAVMsg: 注意，回调结束后，内存块会被PullSession重复使用
func (s *PullSession) Pull(rawURL string, onReadRTMPAVMsg OnReadRTMPAVMsg) error { func (s *PullSession) Pull(rawURL string, onReadRTMPAVMsg OnReadRTMPAVMsg) error {
s.core.onReadRTMPAVMsg = onReadRTMPAVMsg s.core.onReadRTMPAVMsg = onReadRTMPAVMsg
return s.core.DoWithTimeout(rawURL) return s.core.Do(rawURL)
} }
func (s *PullSession) Done() <-chan error { // Pull成功后，调用该函数，可阻塞直到拉流结束
return s.core.Done() func (s *PullSession) Wait() <-chan error {
return s.core.Wait()
} }
func (s *PullSession) Dispose() { func (s *PullSession) Dispose() {

21
pkg/rtmp/client_push_session.go
Unescape Escape View File

@ -8,7 +8,9 @@
package rtmp package rtmp
import "github.com/q191201771/lal/pkg/base" import (
"github.com/q191201771/lal/pkg/base"
)
type PushSession struct { type PushSession struct {
IsFresh bool IsFresh bool
@ -17,13 +19,14 @@ type PushSession struct {
} }
type PushSessionOption struct { type PushSessionOption struct {
ConnectTimeoutMS int // 从调用Push函数，到可以发送音视频数据的前一步，也即收到服务端返回的rtmp publish对应结果的信令的超时时间
PushTimeoutMS int // 如果为0，则没有超时时间
PushTimeoutMS int
WriteAVTimeoutMS int WriteAVTimeoutMS int
} }
var defaultPushSessionOption = PushSessionOption{ var defaultPushSessionOption = PushSessionOption{
ConnectTimeoutMS: 0,
PushTimeoutMS: 0, PushTimeoutMS: 0,
WriteAVTimeoutMS: 0, WriteAVTimeoutMS: 0,
} }
@ -38,17 +41,15 @@ func NewPushSession(modOptions ...ModPushSessionOption) *PushSession {
return &PushSession{ return &PushSession{
IsFresh: true, IsFresh: true,
core: NewClientSession(CSTPushSession, func(option *ClientSessionOption) { core: NewClientSession(CSTPushSession, func(option *ClientSessionOption) {
option.ConnectTimeoutMS = opt.ConnectTimeoutMS
option.DoTimeoutMS = opt.PushTimeoutMS option.DoTimeoutMS = opt.PushTimeoutMS
option.WriteAVTimeoutMS = opt.WriteAVTimeoutMS option.WriteAVTimeoutMS = opt.WriteAVTimeoutMS
}), }),
} }
} }
// 建立rtmp publish连接 // 如果没有错误发生，阻塞到接收音视频数据的前一步，也即收到服务端返回的rtmp publish对应结果的信令
// 阻塞直到收到服务端返回的rtmp publish对应结果的信令，或发生错误
func (s *PushSession) Push(rawURL string) error { func (s *PushSession) Push(rawURL string) error {
return s.core.DoWithTimeout(rawURL) return s.core.Do(rawURL)
} }
func (s *PushSession) AsyncWrite(msg []byte) error { func (s *PushSession) AsyncWrite(msg []byte) error {
@ -87,8 +88,8 @@ func (s *PushSession) RawQuery() string {
return s.core.RawQuery() return s.core.RawQuery()
} }
func (s *PushSession) Done() <-chan error { func (s *PushSession) Wait() <-chan error {
return s.core.Done() return s.core.Wait()
} }
func (s *PushSession) UniqueKey() string { func (s *PushSession) UniqueKey() string {

101
pkg/rtmp/client_session.go
Unescape Escape View File

@ -9,6 +9,7 @@
package rtmp package rtmp
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"net" "net"
@ -56,14 +57,12 @@ const (
type ClientSessionOption struct { type ClientSessionOption struct {
// 单位毫秒，如果为0，则没有超时 // 单位毫秒，如果为0，则没有超时
ConnectTimeoutMS int // 建立连接超时
DoTimeoutMS int // 从发起连接（包含了建立连接的时间）到收到publish或play信令结果的超时 DoTimeoutMS int // 从发起连接（包含了建立连接的时间）到收到publish或play信令结果的超时
ReadAVTimeoutMS int // 读取音视频数据的超时 ReadAVTimeoutMS int // 读取音视频数据的超时
WriteAVTimeoutMS int // 发送音视频数据的超时 WriteAVTimeoutMS int // 发送音视频数据的超时
} }
var defaultClientSessOption = ClientSessionOption{ var defaultClientSessOption = ClientSessionOption{
ConnectTimeoutMS: 0,
DoTimeoutMS: 0, DoTimeoutMS: 0,
ReadAVTimeoutMS: 0, ReadAVTimeoutMS: 0,
WriteAVTimeoutMS: 0, WriteAVTimeoutMS: 0,
@ -103,7 +102,23 @@ func NewClientSession(t ClientSessionType, modOptions ...ModClientSessionOption)
return s return s
} }
func (s *ClientSession) Done() <-chan error { // 阻塞直到收到服务端返回的 publish / play 对应结果的信令或者发生错误
func (s *ClientSession) Do(rawURL string) error {
var (
ctx context.Context
cancel context.CancelFunc
)
if s.option.DoTimeoutMS == 0 {
ctx, cancel = context.WithCancel(context.Background())
} else {
ctx, cancel = context.WithTimeout(context.Background(), time.Duration(s.option.DoTimeoutMS)*time.Millisecond)
}
defer cancel()
return s.doContext(ctx, rawURL)
}
// Do成功后，调用该函数，可阻塞直到推流或拉流结束
func (s *ClientSession) Wait() <-chan error {
return s.conn.Done() return s.conn.Done()
} }
@ -169,62 +184,45 @@ func (s *ClientSession) IsAlive() (readAlive, writeAlive bool) {
return return
} }
// 阻塞直到收到服务端返回的 publish / play 对应结果的信令或者发生错误 func (s *ClientSession) doContext(ctx context.Context, rawURL string) error {
func (s *ClientSession) DoWithTimeout(rawURL string) error { errChan := make(chan error, 1)
if s.option.DoTimeoutMS == 0 {
err := <-s.do(rawURL)
return err
}
t := time.NewTimer(time.Duration(s.option.DoTimeoutMS) * time.Millisecond)
defer t.Stop()
select {
// TODO chef: 这种写法执行不到超时
case err := <-s.do(rawURL):
return err
case <-t.C:
return ErrClientSessionTimeout
}
}
func (s *ClientSession) do(rawURL string) <-chan error { go func() {
ch := make(chan error, 1) if err := s.parseURL(rawURL); err != nil {
if err := s.parseURL(rawURL); err != nil { errChan <- err
ch <- err return
return ch }
} if err := s.tcpConnect(); err != nil {
if err := s.tcpConnect(); err != nil { errChan <- err
ch <- err return
return ch }
}
if err := s.handshake(); err != nil { if err := s.handshake(); err != nil {
ch <- err errChan <- err
return ch return
} }
log.Infof("[%s] > W SetChunkSize %d.", s.UniqueKey, LocalChunkSize) log.Infof("[%s] > W SetChunkSize %d.", s.UniqueKey, LocalChunkSize)
if err := s.packer.writeChunkSize(s.conn, LocalChunkSize); err != nil { if err := s.packer.writeChunkSize(s.conn, LocalChunkSize); err != nil {
ch <- err errChan <- err
return ch return
} }
log.Infof("[%s] > W connect('%s').", s.UniqueKey, s.appName()) log.Infof("[%s] > W connect('%s').", s.UniqueKey, s.appName())
if err := s.packer.writeConnect(s.conn, s.appName(), s.tcURL(), s.t == CSTPushSession); err != nil { if err := s.packer.writeConnect(s.conn, s.appName(), s.tcURL(), s.t == CSTPushSession); err != nil {
ch <- err errChan <- err
return ch return
} }
go s.runReadLoop() s.runReadLoop()
}()
select { select {
case <-ctx.Done():
return ctx.Err()
case <-s.doResultChan: case <-s.doResultChan:
ch <- nil return nil
break
case err := <-s.conn.Done():
ch <- err
break
} }
return ch
} }
func (s *ClientSession) parseURL(rawURL string) (err error) { func (s *ClientSession) parseURL(rawURL string) (err error) {
@ -256,7 +254,7 @@ func (s *ClientSession) tcpConnect() error {
s.stat.RemoteAddr = s.urlCtx.HostWithPort s.stat.RemoteAddr = s.urlCtx.HostWithPort
var conn net.Conn var conn net.Conn
if conn, err = net.DialTimeout("tcp", s.urlCtx.HostWithPort, time.Duration(s.option.ConnectTimeoutMS)*time.Millisecond); err != nil { if conn, err = net.Dial("tcp", s.urlCtx.HostWithPort); err != nil {
return err return err
} }
@ -286,6 +284,7 @@ func (s *ClientSession) handshake() error {
} }
func (s *ClientSession) runReadLoop() { func (s *ClientSession) runReadLoop() {
// TODO chef: 这里是否应该主动关闭conn，考虑对端发送非法协议数据，增加一个对应的测试看看
_ = s.chunkComposer.RunLoop(s.conn, s.doMsg) _ = s.chunkComposer.RunLoop(s.conn, s.doMsg)
} }

8
pkg/rtmp/handshake.go
Unescape Escape View File

@ -100,6 +100,14 @@ func (c *HandshakeClientSimple) WriteC0C1(writer io.Writer) error {
c.c0c1 = make([]byte, c0c1Len) c.c0c1 = make([]byte, c0c1Len)
c.c0c1[0] = version c.c0c1[0] = version
bele.BEPutUint32(c.c0c1[1:5], uint32(time.Now().UnixNano())) bele.BEPutUint32(c.c0c1[1:5], uint32(time.Now().UnixNano()))
//c.c0c1[1] = 0
//c.c0c1[2] = 0
//c.c0c1[3] = 0
//c.c0c1[4] = 0
//c.c0c1[5] = 9
//c.c0c1[6] = 0
//c.c0c1[7] = 124
//c.c0c1[8] = 2
random1528(c.c0c1[9:]) random1528(c.c0c1[9:])
_, err := writer.Write(c.c0c1) _, err := writer.Write(c.c0c1)

18
pkg/rtprtcp/rtp_unpacker.go
Unescape Escape View File

@ -111,18 +111,14 @@ func (r *RTPUnpacker) isStale(seq uint16) bool {
// 计算rtp包处于帧中的位置 // 计算rtp包处于帧中的位置
func (r *RTPUnpacker) calcPositionIfNeeded(pkt *RTPPacket) { func (r *RTPUnpacker) calcPositionIfNeeded(pkt *RTPPacket) {
switch pkt.Header.PacketType { switch r.payloadType {
case base.RTPPacketTypeAVCOrHEVC: case base.AVPacketPTAVC:
switch r.payloadType { calcPositionIfNeededAVC(pkt)
case base.AVPacketPTAVC: case base.AVPacketPTHEVC:
calcPositionIfNeededAVC(pkt) calcPositionIfNeededHEVC(pkt)
case base.AVPacketPTHEVC: case base.AVPacketPTAAC:
calcPositionIfNeededHEVC(pkt)
default:
// can't reach here
}
case base.RTPPacketTypeAAC:
// noop // noop
break
default: default:
// can't reach here // can't reach here
} }

10
pkg/rtprtcp/rtp_unpacker_hevc.go
Unescape Escape View File

@ -25,9 +25,17 @@ func calcPositionIfNeededHEVC(pkt *RTPPacket) {
outerNALUType := hevc.ParseNALUType(b[0]) outerNALUType := hevc.ParseNALUType(b[0])
switch outerNALUType { switch outerNALUType {
case hevc.NALUTypeSliceIDRNLP: case hevc.NALUTypeVPS:
fallthrough
case hevc.NALUTypeSPS:
fallthrough
case hevc.NALUTypePPS:
fallthrough
case hevc.NALUTypeSEI:
fallthrough fallthrough
case hevc.NALUTypeSliceTrailR: case hevc.NALUTypeSliceTrailR:
fallthrough
case hevc.NALUTypeSliceIDRNLP:
pkt.positionType = PositionTypeSingle pkt.positionType = PositionTypeSingle
return return
case NALUTypeHEVCFUA: case NALUTypeHEVCFUA:

85
pkg/rtsp/auth.go
Unescape Escape View File

@ -0,0 +1,85 @@
// Copyright 2020, Chef. All rights reserved.
// https://github.com/q191201771/lal
//
// Use of this source code is governed by a MIT-style license
// that can be found in the License file.
//
// Author: Chef (191201771@qq.com)
package rtsp
import (
"fmt"
"strings"
"github.com/q191201771/naza/pkg/nazalog"
"github.com/q191201771/naza/pkg/nazamd5"
)
// TODO chef: 考虑部分内容移入naza中
// TODO chef: 只支持Digest方式，不支持Basic方式
const (
AuthTypeDigest = "Digest"
AuthAlgorithm = "MD5"
)
type Auth struct {
Username string
Password string
Typ string
Realm string
Nonce string
Algorithm string
}
func (a *Auth) FeedWWWAuthenticate(s, username, password string) {
a.Username = username
a.Password = password
s = strings.TrimPrefix(s, HeaderWWWAuthenticate)
s = strings.TrimSpace(s)
if strings.HasPrefix(s, AuthTypeDigest) {
a.Typ = AuthTypeDigest
}
a.Realm = a.getV(s, `realm="`)
a.Nonce = a.getV(s, `nonce="`)
a.Algorithm = a.getV(s, `algorithm="`)
if a.Typ != AuthTypeDigest {
nazalog.Warnf("FeedWWWAuthenticate type invalid, only support Digest. v=%s", s)
}
if a.Realm == "" {
nazalog.Warnf("FeedWWWAuthenticate realm invalid. v=%s", s)
}
if a.Nonce == "" {
nazalog.Warnf("FeedWWWAuthenticate realm invalid. v=%s", s)
}
if a.Algorithm != AuthAlgorithm {
nazalog.Warnf("FeedWWWAuthenticate algorithm invalid, only support MD5. v=%s", s)
}
}
func (a *Auth) MakeAuthorization(method, uri string) string {
if a.Username == "" || a.Nonce == "" {
return ""
}
ha1 := nazamd5.MD5([]byte(fmt.Sprintf("%s:%s:%s", a.Username, a.Realm, a.Password)))
ha2 := nazamd5.MD5([]byte(fmt.Sprintf("%s:%s", method, uri)))
response := nazamd5.MD5([]byte(fmt.Sprintf("%s:%s:%s", ha1, a.Nonce, ha2)))
return fmt.Sprintf(`%s username="%s", realm="%s", nonce="%s", uri="%s", response="%s", algorithm="%s"`, a.Typ, a.Username, a.Realm, a.Nonce, uri, response, a.Algorithm)
}
func (a *Auth) getV(s string, pre string) string {
b := strings.Index(s, pre)
if b == -1 {
return ""
}
e := strings.Index(s[b+len(pre):], `"`)
if e == -1 {
return ""
}
return s[b+len(pre) : b+len(pre)+e]
}

16
pkg/rtsp/auth_test.go
Unescape Escape View File

@ -0,0 +1,16 @@
// Copyright 2020, Chef. All rights reserved.
// https://github.com/q191201771/lal
//
// Use of this source code is governed by a MIT-style license
// that can be found in the License file.
//
// Author: Chef (191201771@qq.com)
package rtsp_test
import (
"testing"
)
func TestAuth(t *testing.T) {
}

15
pkg/rtsp/avpacket_queue.go
Unescape Escape View File

@ -11,7 +11,6 @@ package rtsp
import ( import (
"github.com/q191201771/lal/pkg/base" "github.com/q191201771/lal/pkg/base"
"github.com/q191201771/naza/pkg/circularqueue" "github.com/q191201771/naza/pkg/circularqueue"
"github.com/q191201771/naza/pkg/nazalog"
) )
// 处理音频和视频的时间戳： // 处理音频和视频的时间戳：
@ -53,12 +52,15 @@ func (a *AVPacketQueue) Feed(pkt base.AVPacket) {
a.videoBaseTS = int64(pkt.Timestamp) a.videoBaseTS = int64(pkt.Timestamp)
} }
pkt.Timestamp -= uint32(a.videoBaseTS) pkt.Timestamp -= uint32(a.videoBaseTS)
_ = a.videoQueue.PushBack(pkt)
if a.videoQueue.Full() { if a.videoQueue.Full() {
pkt, _ := a.videoQueue.Front()
_, _ = a.videoQueue.PopFront() _, _ = a.videoQueue.PopFront()
nazalog.Warnf("video queue full, drop front packet.") ppkt := pkt.(base.AVPacket)
a.onAVPacket(ppkt)
return
} }
_ = a.videoQueue.PushBack(pkt)
//nazalog.Debugf("AVQ v push. a=%d, v=%d", a.audioQueue.Size(), a.videoQueue.Size()) //nazalog.Debugf("AVQ v push. a=%d, v=%d", a.audioQueue.Size(), a.videoQueue.Size())
case base.AVPacketPTAAC: case base.AVPacketPTAAC:
if a.audioBaseTS == -1 { if a.audioBaseTS == -1 {
@ -66,11 +68,14 @@ func (a *AVPacketQueue) Feed(pkt base.AVPacket) {
} }
pkt.Timestamp -= uint32(a.audioBaseTS) pkt.Timestamp -= uint32(a.audioBaseTS)
_ = a.audioQueue.PushBack(pkt)
if a.audioQueue.Full() { if a.audioQueue.Full() {
pkt, _ := a.audioQueue.Front()
_, _ = a.audioQueue.PopFront() _, _ = a.audioQueue.PopFront()
nazalog.Warnf("audio queue full, drop front packet. a=%d, v=%d", a.audioQueue.Size(), a.videoQueue.Size()) ppkt := pkt.(base.AVPacket)
a.onAVPacket(ppkt)
return
} }
_ = a.audioQueue.PushBack(pkt)
//nazalog.Debugf("AVQ a push. a=%d, v=%d", a.audioQueue.Size(), a.videoQueue.Size()) //nazalog.Debugf("AVQ a push. a=%d, v=%d", a.audioQueue.Size(), a.videoQueue.Size())
} //switch loop } //switch loop

52
pkg/rtsp/base_in_session.go
Unescape Escape View File

@ -82,13 +82,21 @@ func (s *BaseInSession) InitWithSDP(rawSDP []byte, sdpLogicCtx sdp.LogicContext)
s.sdpLogicCtx = sdpLogicCtx s.sdpLogicCtx = sdpLogicCtx
s.m.Unlock() s.m.Unlock()
s.audioUnpacker = rtprtcp.NewRTPUnpacker(s.sdpLogicCtx.AudioPayloadType, s.sdpLogicCtx.AudioClockRate, unpackerItemMaxSize, s.onAVPacketUnpacked) if isSupportType(s.sdpLogicCtx.AudioPayloadType) {
s.videoUnpacker = rtprtcp.NewRTPUnpacker(s.sdpLogicCtx.VideoPayloadType, s.sdpLogicCtx.VideoClockRate, unpackerItemMaxSize, s.onAVPacketUnpacked) s.audioUnpacker = rtprtcp.NewRTPUnpacker(s.sdpLogicCtx.AudioPayloadType, s.sdpLogicCtx.AudioClockRate, unpackerItemMaxSize, s.onAVPacketUnpacked)
} else {
nazalog.Warnf("[%s] audio unpacker not support yet. origin type=%d", s.UniqueKey, s.sdpLogicCtx.AudioPayloadTypeOrigin)
}
if isSupportType(s.sdpLogicCtx.VideoPayloadType) {
s.videoUnpacker = rtprtcp.NewRTPUnpacker(s.sdpLogicCtx.VideoPayloadType, s.sdpLogicCtx.VideoClockRate, unpackerItemMaxSize, s.onAVPacketUnpacked)
} else {
nazalog.Warnf("[%s] video unpacker not support yet. origin type=%d", s.UniqueKey, s.sdpLogicCtx.AudioPayloadTypeOrigin)
}
s.audioRRProducer = rtprtcp.NewRRProducer(s.sdpLogicCtx.AudioClockRate) s.audioRRProducer = rtprtcp.NewRRProducer(s.sdpLogicCtx.AudioClockRate)
s.videoRRProducer = rtprtcp.NewRRProducer(s.sdpLogicCtx.VideoClockRate) s.videoRRProducer = rtprtcp.NewRRProducer(s.sdpLogicCtx.VideoClockRate)
if s.sdpLogicCtx.AudioPayloadType != 0 && s.sdpLogicCtx.VideoPayloadType != 0 { if s.sdpLogicCtx.HasAudio && s.sdpLogicCtx.HasVideo {
s.avPacketQueue = NewAVPacketQueue(s.onAVPacket) s.avPacketQueue = NewAVPacketQueue(s.onAVPacket)
} }
@ -97,9 +105,11 @@ func (s *BaseInSession) InitWithSDP(rawSDP []byte, sdpLogicCtx sdp.LogicContext)
} }
} }
// 如果没有设置回调监听对象，可以通过该函数设置，调用方保证调用该函数发生在调用InitWithSDP之后
func (s *BaseInSession) SetObserver(observer BaseInSessionObserver) { func (s *BaseInSession) SetObserver(observer BaseInSessionObserver) {
s.observer = observer s.observer = observer
// TODO chef: 这里的判断应该去掉
if s.sdpLogicCtx.ASC != nil && s.sdpLogicCtx.SPS != nil { if s.sdpLogicCtx.ASC != nil && s.sdpLogicCtx.SPS != nil {
s.observer.OnAVConfig(s.sdpLogicCtx.ASC, s.sdpLogicCtx.VPS, s.sdpLogicCtx.SPS, s.sdpLogicCtx.PPS) s.observer.OnAVConfig(s.sdpLogicCtx.ASC, s.sdpLogicCtx.VPS, s.sdpLogicCtx.SPS, s.sdpLogicCtx.PPS)
} }
@ -122,7 +132,7 @@ func (s *BaseInSession) SetupWithConn(uri string, rtpConn, rtcpConn *nazanet.UDP
return nil return nil
} }
func (s *BaseInSession) SetupWithChannel(uri string, rtpChannel, rtcpChannel int, remoteAddr string) error { func (s *BaseInSession) SetupWithChannel(uri string, rtpChannel, rtcpChannel int) error {
if strings.HasSuffix(uri, s.sdpLogicCtx.AudioAControl) { if strings.HasSuffix(uri, s.sdpLogicCtx.AudioAControl) {
s.audioRTPChannel = rtpChannel s.audioRTPChannel = rtpChannel
s.audioRTCPChannel = rtcpChannel s.audioRTCPChannel = rtcpChannel
@ -213,6 +223,22 @@ func (s *BaseInSession) IsAlive() (readAlive, writeAlive bool) {
return return
} }
// 发现pull时，需要先给对端发送数据，才能收到数据
func (s *BaseInSession) WriteRTPRTCPDummy() {
if s.videoRTPConn != nil {
_ = s.videoRTPConn.Write(dummyRTPPacket)
}
if s.videoRTCPConn != nil {
_ = s.videoRTCPConn.Write(dummyRTCPPacket)
}
if s.audioRTPConn != nil {
_ = s.audioRTPConn.Write(dummyRTPPacket)
}
if s.audioRTCPConn != nil {
_ = s.audioRTCPConn.Write(dummyRTCPPacket)
}
}
// callback by RTPUnpacker // callback by RTPUnpacker
func (s *BaseInSession) onAVPacketUnpacked(pkt base.AVPacket) { func (s *BaseInSession) onAVPacketUnpacked(pkt base.AVPacket) {
if s.avPacketQueue != nil { if s.avPacketQueue != nil {
@ -311,8 +337,8 @@ func (s *BaseInSession) handleRTPPacket(b []byte) error {
return ErrRTSP return ErrRTSP
} }
packetType := b[1] & 0x7F packetType := int(b[1] & 0x7F)
if packetType != base.RTPPacketTypeAVCOrHEVC && packetType != base.RTPPacketTypeAAC { if packetType != s.sdpLogicCtx.AudioPayloadTypeOrigin && packetType != s.sdpLogicCtx.VideoPayloadTypeOrigin {
return ErrRTSP return ErrRTSP
} }
@ -328,16 +354,22 @@ func (s *BaseInSession) handleRTPPacket(b []byte) error {
pkt.Raw = b pkt.Raw = b
switch packetType { switch packetType {
case base.RTPPacketTypeAVCOrHEVC: case s.sdpLogicCtx.VideoPayloadTypeOrigin:
s.videoSSRC = h.SSRC s.videoSSRC = h.SSRC
s.observer.OnRTPPacket(pkt) s.observer.OnRTPPacket(pkt)
s.videoUnpacker.Feed(pkt)
s.videoRRProducer.FeedRTPPacket(h.Seq) s.videoRRProducer.FeedRTPPacket(h.Seq)
case base.RTPPacketTypeAAC:
if s.videoUnpacker != nil {
s.videoUnpacker.Feed(pkt)
}
case s.sdpLogicCtx.AudioPayloadTypeOrigin:
s.audioSSRC = h.SSRC s.audioSSRC = h.SSRC
s.observer.OnRTPPacket(pkt) s.observer.OnRTPPacket(pkt)
s.audioUnpacker.Feed(pkt)
s.audioRRProducer.FeedRTPPacket(h.Seq) s.audioRRProducer.FeedRTPPacket(h.Seq)
if s.audioUnpacker != nil {
s.audioUnpacker.Feed(pkt)
}
default: default:
// 因为前面已经判断过type了，所以永远不会走到这 // 因为前面已经判断过type了，所以永远不会走到这
} }

231
pkg/rtsp/client_pull_session.go
Unescape Escape View File

@ -9,9 +9,11 @@
package rtsp package rtsp
import ( import (
"bufio"
"context" "context"
"fmt" "fmt"
"net" "net"
"strings"
"time" "time"
"github.com/q191201771/lal/pkg/base" "github.com/q191201771/lal/pkg/base"
@ -24,7 +26,8 @@ import (
) )
const ( const (
pullReadBufSize = 256 pullReadBufSize = 256
writeGetParameterIntervalSec = 10
) )
type PullSessionObserver interface { type PullSessionObserver interface {
@ -32,11 +35,16 @@ type PullSessionObserver interface {
} }
type PullSessionOption struct { type PullSessionOption struct {
PullTimeoutMS int // 从调用Pull函数，到收到rtsp play response（接收音视频数据的前一步）的超时时间 // 从调用Pull函数，到接收音视频数据的前一步，也即收到rtsp play response的超时时间
// 如果为0，则没有超时时间
PullTimeoutMS int
OverTCP bool // 是否使用interleaved模式，也即是否通过rtsp command tcp连接传输rtp/rtcp数据
} }
var defaultPullSessionOption = PullSessionOption{ var defaultPullSessionOption = PullSessionOption{
PullTimeoutMS: 10000, PullTimeoutMS: 10000,
OverTCP: false,
} }
type PullSession struct { type PullSession struct {
@ -48,9 +56,16 @@ type PullSession struct {
cseq int cseq int
sessionID string sessionID string
channel int
rawURL string rawURL string
urlCtx base.URLContext urlCtx base.URLContext
waitErrChan chan error
methodGetParameterSupported bool
auth Auth
} }
type ModPullSessionOption func(option *PullSessionOption) type ModPullSessionOption func(option *PullSessionOption)
@ -63,8 +78,9 @@ func NewPullSession(observer PullSessionObserver, modOptions ...ModPullSessionOp
uk := base.GenUniqueKey(base.UKPRTSPPullSession) uk := base.GenUniqueKey(base.UKPRTSPPullSession)
s := &PullSession{ s := &PullSession{
UniqueKey: uk, UniqueKey: uk,
option: option, option: option,
waitErrChan: make(chan error, 1),
} }
baseInSession := &BaseInSession{ baseInSession := &BaseInSession{
UniqueKey: uk, UniqueKey: uk,
@ -81,6 +97,7 @@ func NewPullSession(observer PullSessionObserver, modOptions ...ModPullSessionOp
return s return s
} }
// 如果没有错误发生，阻塞直到接收音视频数据的前一步，也即收到rtsp play response
func (session *PullSession) Pull(rawURL string) error { func (session *PullSession) Pull(rawURL string) error {
var ( var (
ctx context.Context ctx context.Context
@ -95,9 +112,15 @@ func (session *PullSession) Pull(rawURL string) error {
return session.pullContext(ctx, rawURL) return session.pullContext(ctx, rawURL)
} }
// Pull成功后，调用该函数，可阻塞直到拉流结束
func (session *PullSession) Wait() <-chan error {
return session.waitErrChan
}
func (session *PullSession) Write(channel int, b []byte) error { func (session *PullSession) Write(channel int, b []byte) error {
return nil return nil
} }
func (session *PullSession) Dispose() error { func (session *PullSession) Dispose() error {
return nil return nil
} }
@ -105,9 +128,11 @@ func (session *PullSession) Dispose() error {
func (session *PullSession) AppName() string { func (session *PullSession) AppName() string {
return session.urlCtx.PathWithoutLastItem return session.urlCtx.PathWithoutLastItem
} }
func (session *PullSession) StreamName() string { func (session *PullSession) StreamName() string {
return session.urlCtx.LastItemOfPath return session.urlCtx.LastItemOfPath
} }
func (session *PullSession) RawQuery() string { func (session *PullSession) RawQuery() string {
return session.urlCtx.RawQuery return session.urlCtx.RawQuery
} }
@ -166,9 +191,91 @@ func (session *PullSession) pullContext(ctx context.Context, rawURL string) erro
} }
} }
dummy := make([]byte, 1) // 用于接收TCP对端关闭FIN信号 go session.runReadLoop()
_, err := session.CmdConn.Read(dummy) return nil
return err }
func (session *PullSession) runReadLoop() {
if !session.methodGetParameterSupported {
// TCP模式，需要收取数据进行处理
if session.option.OverTCP {
var r = bufio.NewReader(session.CmdConn)
for {
isInterleaved, packet, channel, err := readInterleaved(r)
if err != nil {
session.waitErrChan <- err
return
}
if isInterleaved {
session.baseInSession.HandleInterleavedPacket(packet, int(channel))
}
}
}
// not over tcp
// 接收TCP对端关闭FIN信号
dummy := make([]byte, 1)
_, err := session.CmdConn.Read(dummy)
session.waitErrChan <- err
return
}
// 对端支持get_parameter，需要定时向对端发送get_parameter进行保活
var r = bufio.NewReader(session.CmdConn)
t := time.NewTicker(writeGetParameterIntervalSec * time.Millisecond)
defer t.Stop()
if session.option.OverTCP {
for {
select {
case <-t.C:
session.cseq++
req := PackRequestGetParameter(session.urlCtx.RawURLWithoutUserInfo, session.cseq, session.sessionID)
if _, err := session.CmdConn.Write([]byte(req)); err != nil {
session.waitErrChan <- err
return
}
default:
// noop
}
isInterleaved, packet, channel, err := readInterleaved(r)
if err != nil {
session.waitErrChan <- err
return
}
if isInterleaved {
session.baseInSession.HandleInterleavedPacket(packet, int(channel))
} else {
if _, err := nazahttp.ReadHTTPResponseMessage(r); err != nil {
session.waitErrChan <- err
return
}
}
}
}
// not over tcp
for {
select {
case <-t.C:
session.cseq++
req := PackRequestGetParameter(session.urlCtx.RawURLWithoutUserInfo, session.cseq, session.sessionID)
if _, err := session.CmdConn.Write([]byte(req)); err != nil {
session.waitErrChan <- err
return
}
if _, err := nazahttp.ReadHTTPResponseMessage(r); err != nil {
session.waitErrChan <- err
return
}
default:
// noop
}
}
} }
func (session *PullSession) connect(rawURL string) (err error) { func (session *PullSession) connect(rawURL string) (err error) {
@ -195,7 +302,7 @@ func (session *PullSession) connect(rawURL string) (err error) {
func (session *PullSession) writeOptions() error { func (session *PullSession) writeOptions() error {
session.cseq++ session.cseq++
req := PackRequestOptions(session.rawURL, session.cseq) req := PackRequestOptions(session.urlCtx.RawURLWithoutUserInfo, session.cseq, "")
nazalog.Debugf("[%s] > write options.", session.UniqueKey) nazalog.Debugf("[%s] > write options.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil { if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err return err
@ -205,12 +312,19 @@ func (session *PullSession) writeOptions() error {
return err return err
} }
nazalog.Debugf("[%s] < read response. %s", session.UniqueKey, ctx.StatusCode) nazalog.Debugf("[%s] < read response. %s", session.UniqueKey, ctx.StatusCode)
session.handleOptionMethods(ctx)
if err := session.handleAuth(ctx); err != nil {
return err
}
return nil return nil
} }
func (session *PullSession) writeDescribe() error { func (session *PullSession) writeDescribe() error {
session.cseq++ session.cseq++
req := PackRequestDescribe(session.rawURL, session.cseq) auth := session.auth.MakeAuthorization(MethodDescribe, session.urlCtx.RawURLWithoutUserInfo)
req := PackRequestDescribe(session.urlCtx.RawURLWithoutUserInfo, session.cseq, auth)
nazalog.Debugf("[%s] > write describe.", session.UniqueKey) nazalog.Debugf("[%s] > write describe.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil { if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err return err
@ -221,39 +335,52 @@ func (session *PullSession) writeDescribe() error {
} }
nazalog.Debugf("[%s] < read response. code=%s, body=%s", session.UniqueKey, ctx.StatusCode, string(ctx.Body)) nazalog.Debugf("[%s] < read response. code=%s, body=%s", session.UniqueKey, ctx.StatusCode, string(ctx.Body))
sdpCtx, err := sdp.ParseSDP2LogicContext(ctx.Body) sdpLogicCtx, err := sdp.ParseSDP2LogicContext(ctx.Body)
if err != nil { if err != nil {
return err return err
} }
session.baseInSession.InitWithSDP(ctx.Body, sdpCtx) session.baseInSession.InitWithSDP(ctx.Body, sdpLogicCtx)
return nil return nil
} }
func (session *PullSession) writeSetup() error { func (session *PullSession) writeSetup() error {
if session.baseInSession.sdpLogicCtx.VideoAControl != "" { if session.baseInSession.sdpLogicCtx.VideoAControl != "" {
if err := session.writeOneSetup(session.baseInSession.sdpLogicCtx.VideoAControl); err != nil { if session.option.OverTCP {
return err if err := session.writeOneSetupTCP(session.baseInSession.sdpLogicCtx.VideoAControl); err != nil {
return err
}
} else {
if err := session.writeOneSetup(session.baseInSession.sdpLogicCtx.VideoAControl); err != nil {
return err
}
} }
} }
if session.baseInSession.sdpLogicCtx.AudioAControl != "" { if session.baseInSession.sdpLogicCtx.AudioAControl != "" {
if err := session.writeOneSetup(session.baseInSession.sdpLogicCtx.AudioAControl); err != nil { if session.option.OverTCP {
return err if err := session.writeOneSetupTCP(session.baseInSession.sdpLogicCtx.AudioAControl); err != nil {
return err
}
} else {
if err := session.writeOneSetup(session.baseInSession.sdpLogicCtx.AudioAControl); err != nil {
return err
}
} }
} }
return nil return nil
} }
func (session *PullSession) writeOneSetup(aControl string) error { func (session *PullSession) writeOneSetup(aControl string) error {
setupURI := fmt.Sprintf("%s/%s", session.rawURL, aControl) setupURI := makeSetupURI(session.urlCtx, aControl)
rtpC, rtpPort, rtcpC, rtcpPort, err := availUDPConnPool.Acquire2() rtpC, rtpPort, rtcpC, rtcpPort, err := availUDPConnPool.Acquire2()
if err != nil { if err != nil {
return err return err
} }
session.cseq++ session.cseq++
req := PackRequestSetup(setupURI, session.cseq, session.sessionID, int(rtpPort), int(rtcpPort)) auth := session.auth.MakeAuthorization(MethodSetup, session.urlCtx.RawURLWithoutUserInfo)
req := PackRequestSetup(setupURI, session.cseq, int(rtpPort), int(rtcpPort), session.sessionID, auth)
nazalog.Debugf("[%s] > write setup.", session.UniqueKey) nazalog.Debugf("[%s] > write setup.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil { if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err return err
@ -264,7 +391,7 @@ func (session *PullSession) writeOneSetup(aControl string) error {
} }
nazalog.Debugf("[%s] < read response. code=%s, ctx=%+v", session.UniqueKey, ctx.StatusCode, ctx) nazalog.Debugf("[%s] < read response. code=%s, ctx=%+v", session.UniqueKey, ctx.StatusCode, ctx)
session.sessionID = ctx.Headers[HeaderFieldSession] session.sessionID = strings.Split(ctx.Headers[HeaderFieldSession], ";")[0]
srvRTPPort, srvRTCPPort, err := parseServerPort(ctx.Headers[HeaderFieldTransport]) srvRTPPort, srvRTCPPort, err := parseServerPort(ctx.Headers[HeaderFieldTransport])
if err != nil { if err != nil {
@ -296,9 +423,42 @@ func (session *PullSession) writeOneSetup(aControl string) error {
return nil return nil
} }
func (session *PullSession) writeOneSetupTCP(aControl string) error {
setupURI := makeSetupURI(session.urlCtx, aControl)
rtpChannel := session.channel
rtcpChannel := session.channel + 1
session.channel += 2
session.cseq++
auth := session.auth.MakeAuthorization(MethodSetup, session.urlCtx.RawURLWithoutUserInfo)
req := PackRequestSetupTCP(setupURI, session.cseq, rtpChannel, rtcpChannel, session.sessionID, auth)
nazalog.Debugf("[%s] > write setup.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err
}
ctx, err := nazahttp.ReadHTTPResponseMessage(session.CmdConn)
if err != nil {
return err
}
nazalog.Debugf("[%s] < read response. code=%s, ctx=%+v", session.UniqueKey, ctx.StatusCode, ctx)
session.sessionID = strings.Split(ctx.Headers[HeaderFieldSession], ";")[0]
// TODO chef: 这里没有解析回传的channel id了，因为我假定了它和request中的是一致的
if err := session.baseInSession.SetupWithChannel(setupURI, rtpChannel, rtcpChannel); err != nil {
return err
}
return nil
}
func (session *PullSession) writePlay() error { func (session *PullSession) writePlay() error {
session.baseInSession.WriteRTPRTCPDummy()
session.cseq++ session.cseq++
req := PackRequestPlay(session.rawURL, session.cseq, session.sessionID) auth := session.auth.MakeAuthorization(MethodPlay, session.urlCtx.RawURLWithoutUserInfo)
req := PackRequestPlay(session.urlCtx.RawURLWithoutUserInfo, session.cseq, session.sessionID, auth)
nazalog.Debugf("[%s] > write play.", session.UniqueKey) nazalog.Debugf("[%s] > write play.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil { if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err return err
@ -310,3 +470,36 @@ func (session *PullSession) writePlay() error {
nazalog.Debugf("[%s] < read response. %s", session.UniqueKey, ctx.StatusCode) nazalog.Debugf("[%s] < read response. %s", session.UniqueKey, ctx.StatusCode)
return nil return nil
} }
func (session *PullSession) handleOptionMethods(ctx nazahttp.HTTPRespMsgCtx) {
methods := ctx.Headers["Public"]
if methods == "" {
return
}
if strings.Contains(methods, MethodGetParameter) {
session.methodGetParameterSupported = true
}
}
func (session *PullSession) handleAuth(ctx nazahttp.HTTPRespMsgCtx) error {
if ctx.Headers[HeaderWWWAuthenticate] == "" {
return nil
}
session.auth.FeedWWWAuthenticate(ctx.Headers[HeaderWWWAuthenticate], session.urlCtx.Username, session.urlCtx.Password)
auth := session.auth.MakeAuthorization(MethodOptions, session.urlCtx.RawURLWithoutUserInfo)
session.cseq++
req := PackRequestOptions(session.urlCtx.RawURLWithoutUserInfo, session.cseq, auth)
nazalog.Debugf("[%s] > write options.", session.UniqueKey)
if _, err := session.CmdConn.Write([]byte(req)); err != nil {
return err
}
ctx, err := nazahttp.ReadHTTPResponseMessage(session.CmdConn)
if err != nil {
return err
}
nazalog.Debugf("[%s] < read response. %s", session.UniqueKey, ctx.StatusCode)
return nil
}

132
pkg/rtsp/pack.go
Unescape Escape View File

@ -16,11 +16,6 @@ import (
) )
// rfc2326 10.1 OPTIONS // rfc2326 10.1 OPTIONS
// uri CSeq
var RequestOptionsTmpl = "OPTIONS %s RTSP/1.0\r\n" +
"CSeq: %d\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// CSeq // CSeq
var ResponseOptionsTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseOptionsTmpl = "RTSP/1.0 200 OK\r\n" +
@ -30,18 +25,14 @@ var ResponseOptionsTmpl = "RTSP/1.0 200 OK\r\n" +
"\r\n" "\r\n"
// rfc2326 10.3 ANNOUNCE // rfc2326 10.3 ANNOUNCE
//var RequestAnnounceTmpl = "not impl"
// CSeq // CSeq
var ResponseAnnounceTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseAnnounceTmpl = "RTSP/1.0 200 OK\r\n" +
"CSeq: %s\r\n" + "CSeq: %s\r\n" +
"\r\n" "\r\n"
// rfc2326 10.2 DESCRIBE // rfc2326 10.2 DESCRIBE
// uri CSeq
var RequestDescribeTmpl = "DESCRIBE %s RTSP/1.0\r\n" +
"Accept: application/sdp\r\n" +
"CSeq: %d\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// CSeq, Date, Content-Length, // CSeq, Date, Content-Length,
var ResponseDescribeTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseDescribeTmpl = "RTSP/1.0 200 OK\r\n" +
@ -53,20 +44,7 @@ var ResponseDescribeTmpl = "RTSP/1.0 200 OK\r\n" +
"%s" "%s"
// rfc2326 10.4 SETUP // rfc2326 10.4 SETUP
// uri CSeq RTPPort RTCPPort // TODO chef: mode=record，这个是咋作用，是应该pub有sub没有吗，我的pack实现没有严格区分
var RequestSetupTmpl = "SETUP %s RTSP/1.0\r\n" +
"CSeq: %d\r\n" +
"Transport: RTP/AVP/UDP;unicast;client_port=%d-%d\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// uri CSeq Session RTPPort RTCPPort
var RequestSetupWithSessionTmpl = "SETUP %s RTSP/1.0\r\n" +
"CSeq: %d\r\n" +
"Session: %s\r\n" +
"Transport: RTP/AVP/UDP;unicast;client_port=%d-%d\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// CSeq, Date, Session, Transport(client_port, server_rtp_port, server_rtcp_port) // CSeq, Date, Session, Transport(client_port, server_rtp_port, server_rtcp_port)
var ResponseSetupTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseSetupTmpl = "RTSP/1.0 200 OK\r\n" +
@ -85,6 +63,8 @@ var ResponseSetupTCPTmpl = "RTSP/1.0 200 OK\r\n" +
"\r\n" "\r\n"
// rfc2326 10.11 RECORD // rfc2326 10.11 RECORD
//var RequestRecordTmpl = "not impl"
// CSeq, Session // CSeq, Session
var ResponseRecordTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseRecordTmpl = "RTSP/1.0 200 OK\r\n" +
"CSeq: %s\r\n" + "CSeq: %s\r\n" +
@ -92,13 +72,6 @@ var ResponseRecordTmpl = "RTSP/1.0 200 OK\r\n" +
"\r\n" "\r\n"
// rfc2326 10.5 PLAY // rfc2326 10.5 PLAY
// uri CSeq Session
var RequestPlayTmpl = "PLAY %s RTSP/1.0\r\n" +
"CSeq: %d\r\n" +
"Range: npt=0.000-\r\n" +
"Session: %s\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// CSeq Date // CSeq Date
var ResponsePlayTmpl = "RTSP/1.0 200 OK\r\n" + var ResponsePlayTmpl = "RTSP/1.0 200 OK\r\n" +
@ -106,30 +79,96 @@ var ResponsePlayTmpl = "RTSP/1.0 200 OK\r\n" +
"Date: %s\r\n" + "Date: %s\r\n" +
"\r\n" "\r\n"
// rfc2326 10.8 GET_PARAMETER
// uri CSeq Session
var RequestGetParameterTmpl = "GET_PARAMETER %s RTSP/1.0\r\n" +
"CSeq: %d\r\n" +
"Session: %s\r\n" +
"User-Agent: " + base.LALRTSPPullSessionUA + "\r\n" +
"\r\n"
// rfc2326 10.7 TEARDOWN // rfc2326 10.7 TEARDOWN
//var RequestTeardownTmpl = "not impl"
// CSeq // CSeq
var ResponseTeardownTmpl = "RTSP/1.0 200 OK\r\n" + var ResponseTeardownTmpl = "RTSP/1.0 200 OK\r\n" +
"CSeq: %s\r\n" + "CSeq: %s\r\n" +
"\r\n" "\r\n"
func PackRequestOptions(uri string, cseq int) string { // @param auth 可以为空，如果为空，则请求中不包含`Authorization`字段
return fmt.Sprintf(RequestOptionsTmpl, uri, cseq) func PackRequestOptions(uri string, cseq int, auth string) string {
headers := map[string]string{
HeaderFieldCSeq: fmt.Sprintf("%d", cseq),
HeaderUserAgent: base.LALRTSPPullSessionUA,
}
if auth != "" {
headers[HeaderAuthorization] = auth
}
return packRequest(MethodOptions, uri, headers)
} }
func PackRequestDescribe(uri string, cseq int) string { // @param auth 可以为空，如果为空，则请求中不包含`Authorization`字段
return fmt.Sprintf(RequestDescribeTmpl, uri, cseq) func PackRequestDescribe(uri string, cseq int, auth string) string {
headers := map[string]string{
HeaderAccept: "application/sdp",
HeaderFieldCSeq: fmt.Sprintf("%d", cseq),
HeaderUserAgent: base.LALRTSPPullSessionUA,
}
if auth != "" {
headers[HeaderAuthorization] = auth
}
return packRequest(MethodDescribe, uri, headers)
} }
// @param sessionID 可以为空，如果为空，则请求中不包含`Session`字段 // @param sessionID 可以为空，如果为空，则请求中不包含`Session`字段
func PackRequestSetup(uri string, cseq int, sessionID string, rtpClientPort int, rtcpClientPort int) string { // @param auth 可以为空，如果为空，则请求中不包含`Authorization`字段
if sessionID == "" { func PackRequestSetup(uri string, cseq int, rtpClientPort int, rtcpClientPort int, sessionID string, auth string) string {
return fmt.Sprintf(RequestSetupTmpl, uri, cseq, rtpClientPort, rtcpClientPort) headers := map[string]string{
HeaderFieldTransport: fmt.Sprintf("RTP/AVP/UDP;unicast;client_port=%d-%d", rtpClientPort, rtcpClientPort),
HeaderFieldCSeq: fmt.Sprintf("%d", cseq),
HeaderUserAgent: base.LALRTSPPullSessionUA,
}
if sessionID != "" {
headers[HeaderFieldSession] = sessionID
} }
return fmt.Sprintf(RequestSetupWithSessionTmpl, uri, cseq, sessionID, rtpClientPort, rtcpClientPort) if auth != "" {
headers[HeaderAuthorization] = auth
}
return packRequest(MethodSetup, uri, headers)
} }
func PackRequestPlay(uri string, cseq int, sessionID string) string { // @param sessionID 可以为空，如果为空，则请求中不包含`Session`字段
return fmt.Sprintf(RequestPlayTmpl, uri, cseq, sessionID) // @param auth 可以为空，如果为空，则请求中不包含`Authorization`字段
func PackRequestSetupTCP(uri string, cseq int, rtpChannel int, rtcpChannel int, sessionID string, auth string) string {
headers := map[string]string{
HeaderFieldCSeq: fmt.Sprintf("%d", cseq),
HeaderFieldTransport: fmt.Sprintf("RTP/AVP/TCP;unicast;interleaved=%d-%d", rtpChannel, rtcpChannel),
HeaderUserAgent: base.LALRTSPPullSessionUA,
}
if sessionID != "" {
headers[HeaderFieldSession] = sessionID
}
if auth != "" {
headers[HeaderAuthorization] = auth
}
return packRequest(MethodSetup, uri, headers)
}
func PackRequestPlay(uri string, cseq int, sessionID string, auth string) string {
headers := map[string]string{
HeaderFieldCSeq: fmt.Sprintf("%d", cseq),
HeaderFieldRange: "npt=0.000-",
HeaderFieldSession: sessionID,
HeaderUserAgent: base.LALRTSPPullSessionUA,
}
if auth != "" {
headers[HeaderAuthorization] = auth
}
return packRequest(MethodPlay, uri, headers)
}
func PackRequestGetParameter(uri string, cseq int, sessionID string) string {
return fmt.Sprintf(RequestGetParameterTmpl, uri, cseq, sessionID)
} }
func PackResponseOptions(cseq string) string { func PackResponseOptions(cseq string) string {
@ -175,3 +214,12 @@ func PackResponsePlay(cseq string) string {
func PackResponseTeardown(cseq string) string { func PackResponseTeardown(cseq string) string {
return fmt.Sprintf(ResponseTeardownTmpl, cseq) return fmt.Sprintf(ResponseTeardownTmpl, cseq)
} }
func packRequest(method, uri string, headers map[string]string) (ret string) {
ret = method + " " + uri + " RTSP/1.0\r\n"
for k, v := range headers {
ret += k + ": " + v + "\r\n"
}
ret += "\r\n"
return ret
}

73
pkg/rtsp/rtsp.go
Unescape Escape View File

@ -15,6 +15,8 @@ import (
"strconv" "strconv"
"strings" "strings"
"github.com/q191201771/lal/pkg/base"
"github.com/q191201771/lal/pkg/rtprtcp" "github.com/q191201771/lal/pkg/rtprtcp"
"github.com/q191201771/naza/pkg/nazalog" "github.com/q191201771/naza/pkg/nazalog"
@ -29,25 +31,30 @@ import (
// - stat // - stat
// - pub和sub存在一些重复代码 // - pub和sub存在一些重复代码
// - sub缺少主动发送sr // - sub缺少主动发送sr
// - queue的策略，当一条流没数据之后 // - pull session回调有observer interface和on func回调两种方式，是否需要统一
// - 用context重写其它pull session
var ErrRTSP = errors.New("lal.rtsp: fxxk") var ErrRTSP = errors.New("lal.rtsp: fxxk")
const ( const (
MethodOptions = "OPTIONS" MethodOptions = "OPTIONS"
MethodAnnounce = "ANNOUNCE" MethodAnnounce = "ANNOUNCE"
MethodDescribe = "DESCRIBE" MethodDescribe = "DESCRIBE"
MethodSetup = "SETUP" MethodSetup = "SETUP"
MethodRecord = "RECORD" MethodRecord = "RECORD"
MethodPlay = "PLAY" MethodPlay = "PLAY"
MethodTeardown = "TEARDOWN" MethodTeardown = "TEARDOWN"
MethodGetParameter = "GET_PARAMETER"
) )
const ( const (
HeaderFieldCSeq = "CSeq" HeaderAccept = "Accept"
HeaderFieldTransport = "Transport" HeaderUserAgent = "User-Agent"
HeaderFieldSession = "Session" HeaderFieldCSeq = "CSeq"
HeaderFieldTransport = "Transport"
HeaderFieldSession = "Session"
HeaderFieldRange = "Range"
HeaderWWWAuthenticate = "WWW-Authenticate"
HeaderAuthorization = "Authorization"
) )
const ( const (
@ -64,12 +71,23 @@ var (
// TODO chef: 参考协议标准，不要使用固定值 // TODO chef: 参考协议标准，不要使用固定值
sessionID = "191201771" sessionID = "191201771"
minServerPort = uint16(8000) minServerPort = uint16(30000)
maxServerPort = uint16(16000) maxServerPort = uint16(60000)
unpackerItemMaxSize = 1024 unpackerItemMaxSize = 1024
serverCommandSessionReadBufSize = 256 serverCommandSessionReadBufSize = 256
dummyRTPPacket = []byte{
0x80, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00,
}
dummyRTCPPacket = []byte{
0x80, 0xc9, 0x00, 0x01,
0x00, 0x00, 0x00, 0x00,
}
) )
var availUDPConnPool *nazanet.AvailUDPConnPool var availUDPConnPool *nazanet.AvailUDPConnPool
@ -148,6 +166,25 @@ func parseTransport(setupTransport string, key string) (first, second uint16, er
return uint16(iFirst), uint16(iSecond), err return uint16(iFirst), uint16(iSecond), err
} }
func isSupportType(t base.AVPacketPT) bool {
switch t {
case base.AVPacketPTAAC:
fallthrough
case base.AVPacketPTAVC:
fallthrough
case base.AVPacketPTHEVC:
return true
}
return false
}
func makeSetupURI(urlCtx base.URLContext, aControl string) string {
if strings.HasPrefix(aControl, "rtsp://") {
return aControl
}
return fmt.Sprintf("%s/%s", urlCtx.RawURLWithoutUserInfo, aControl)
}
func init() { func init() {
availUDPConnPool = nazanet.NewAvailUDPConnPool(minServerPort, maxServerPort) availUDPConnPool = nazanet.NewAvailUDPConnPool(minServerPort, maxServerPort)
} }
@ -177,9 +214,6 @@ func init() {
// read http request. method=SETUP, uri=rtsp://localhost:5544/live/test110/streamid=1, headers=map[CSeq:4 Session:191201771 Transport:RTP/AVP/UDP;unicast;client_port=32184-32185;mode=record User-Agent:Lavf57.83.100], body= - server.go:95 // read http request. method=SETUP, uri=rtsp://localhost:5544/live/test110/streamid=1, headers=map[CSeq:4 Session:191201771 Transport:RTP/AVP/UDP;unicast;client_port=32184-32185;mode=record User-Agent:Lavf57.83.100], body= - server.go:95
// read http request. method=RECORD, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:5 Range:npt=0.000- Session:191201771 User-Agent:Lavf57.83.100], body= - server.go:95 // read http request. method=RECORD, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:5 Range:npt=0.000- Session:191201771 User-Agent:Lavf57.83.100], body= - server.go:95
// read http request. method=TEARDOWN, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:6 Session:191201771 User-Agent:Lavf57.83.100], body= - server.go:95 // read http request. method=TEARDOWN, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:6 Session:191201771 User-Agent:Lavf57.83.100], body= - server.go:95
// read udp packet failed. err=read udp [::]:8002: use of closed network connection - server_pub_session.go:199
// read udp packet failed. err=read udp [::]:8003: use of closed network connection - server_pub_session.go:199
// --------------------------------------------------------------------------------------------------------------------- // ---------------------------------------------------------------------------------------------------------------------
// --------------------------------------------------------------------------------------------------------------------- // ---------------------------------------------------------------------------------------------------------------------
@ -221,6 +255,11 @@ func init() {
// --------------------------------------------------------------------------------------------------------------------- // ---------------------------------------------------------------------------------------------------------------------
// SUB(rtp over tcp) // SUB(rtp over tcp)
// //
// read http request. method=OPTIONS, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:1 User-Agent:Lavf57.83.100], body= - server_command_session.go:136
// read http request. method=DESCRIBE, uri=rtsp://localhost:5544/live/test110, headers=map[Accept:application/sdp CSeq:2 User-Agent:Lavf57.83.100], body= - server_command_session.go:136
// read http request. method=SETUP, uri=rtsp://localhost:5544/live/test110/streamid=0, headers=map[CSeq:3 Transport:RTP/AVP/TCP;unicast;interleaved=0-1 User-Agent:Lavf57.83.100], body= - server_command_session.go:136
// read http request. method=SETUP, uri=rtsp://localhost:5544/live/test110/streamid=1, headers=map[CSeq:4 Session:191201771 Transport:RTP/AVP/TCP;unicast;interleaved=2-3 User-Agent:Lavf57.83.100], body= - server_command_session.go:136
// read http request. method=PLAY, uri=rtsp://localhost:5544/live/test110, headers=map[CSeq:5 Range:npt=0.000- Session:191201771 User-Agent:Lavf57.83.100], body= - server_command_session.go:136
// --------------------------------------------------------------------------------------------------------------------- // ---------------------------------------------------------------------------------------------------------------------
// 8000 video rtp // 8000 video rtp

4
pkg/rtsp/server_command_session.go
Unescape Escape View File

@ -227,8 +227,8 @@ func (s *ServerCommandSession) handleDescribe(requestCtx nazahttp.HTTPReqMsgCtx)
return ErrRTSP return ErrRTSP
} }
ctx, _ := sdp.ParseSDP2LogicContext(rawSDP) sdpLogicCtx, _ := sdp.ParseSDP2LogicContext(rawSDP)
s.subSession.InitWithSDP(rawSDP, ctx) s.subSession.InitWithSDP(rawSDP, sdpLogicCtx)
resp := PackResponseDescribe(requestCtx.Headers[HeaderFieldCSeq], string(rawSDP)) resp := PackResponseDescribe(requestCtx.Headers[HeaderFieldCSeq], string(rawSDP))
_, err = s.conn.Write([]byte(resp)) _, err = s.conn.Write([]byte(resp))

2
pkg/rtsp/server_pub_session.go
Unescape Escape View File

@ -65,7 +65,7 @@ func (s *PubSession) SetupWithConn(uri string, rtpConn, rtcpConn *nazanet.UDPCon
} }
func (s *PubSession) SetupWithChannel(uri string, rtpChannel, rtcpChannel int, remoteAddr string) error { func (s *PubSession) SetupWithChannel(uri string, rtpChannel, rtcpChannel int, remoteAddr string) error {
return s.baseInSession.SetupWithChannel(uri, rtpChannel, rtcpChannel, remoteAddr) return s.baseInSession.SetupWithChannel(uri, rtpChannel, rtcpChannel)
} }
func (s *PubSession) Dispose() { func (s *PubSession) Dispose() {

4
pkg/rtsp/server_sub_session.go
Unescape Escape View File

@ -178,14 +178,14 @@ func (s *SubSession) WriteRTPPacket(packet rtprtcp.RTPPacket) {
atomic.AddUint64(&s.currConnStat.WroteBytesSum, uint64(len(packet.Raw))) atomic.AddUint64(&s.currConnStat.WroteBytesSum, uint64(len(packet.Raw)))
switch packet.Header.PacketType { switch packet.Header.PacketType {
case base.RTPPacketTypeAVCOrHEVC: case uint8(s.sdpLogicCtx.VideoPayloadTypeOrigin):
if s.videoRTPConn != nil { if s.videoRTPConn != nil {
_ = s.videoRTPConn.Write(packet.Raw) _ = s.videoRTPConn.Write(packet.Raw)
} }
if s.videoRTPChannel != -1 { if s.videoRTPChannel != -1 {
_ = s.cmdSession.Write(s.videoRTPChannel, packet.Raw) _ = s.cmdSession.Write(s.videoRTPChannel, packet.Raw)
} }
case base.RTPPacketTypeAAC: case uint8(s.sdpLogicCtx.AudioPayloadTypeOrigin):
if s.audioRTPConn != nil { if s.audioRTPConn != nil {
_ = s.audioRTPConn.Write(packet.Raw) _ = s.audioRTPConn.Write(packet.Raw)
} }

170
pkg/sdp/sdp.go
Unescape Escape View File

@ -22,26 +22,41 @@ import (
var ErrSDP = errors.New("lal.sdp: fxxk") var ErrSDP = errors.New("lal.sdp: fxxk")
const ( const (
ARTPMapEncodingName = "H265" ARTPMapEncodingNameH265 = "H265"
ARTPMapEncodingNameH264 = "H264"
ARTPMapEncodingNameAAC = "MPEG4-GENERIC"
) )
type LogicContext struct { type LogicContext struct {
AudioClockRate int HasAudio bool
VideoClockRate int HasVideo bool
AudioPayloadType base.AVPacketPT AudioClockRate int
VideoPayloadType base.AVPacketPT VideoClockRate int
AudioAControl string AudioPayloadTypeOrigin int
VideoAControl string VideoPayloadTypeOrigin int
ASC []byte AudioPayloadType base.AVPacketPT
VPS []byte VideoPayloadType base.AVPacketPT
SPS []byte AudioAControl string
PPS []byte VideoAControl string
ASC []byte
VPS []byte
SPS []byte
PPS []byte
}
type MediaDesc struct {
M M
ARTPMap ARTPMap
AFmtBase AFmtPBase
AControl AControl
} }
type RawContext struct { type RawContext struct {
ARTPMapList []ARTPMap MediaDescList []MediaDesc
AFmtPBaseList []AFmtPBase }
AControlList []AControl
type M struct {
Media string
} }
type ARTPMap struct { type ARTPMap struct {
@ -68,47 +83,45 @@ func ParseSDP2LogicContext(b []byte) (LogicContext, error) {
return ret, err return ret, err
} }
for i, item := range c.ARTPMapList { for _, md := range c.MediaDescList {
switch item.PayloadType { switch md.M.Media {
case base.RTPPacketTypeAVCOrHEVC: case "audio":
ret.VideoClockRate = item.ClockRate ret.HasAudio = true
if item.EncodingName == ARTPMapEncodingName { ret.AudioClockRate = md.ARTPMap.ClockRate
ret.VideoPayloadType = base.AVPacketPTHEVC ret.AudioAControl = md.AControl.Value
ret.AudioPayloadTypeOrigin = md.ARTPMap.PayloadType
if md.ARTPMap.EncodingName == ARTPMapEncodingNameAAC {
ret.AudioPayloadType = base.AVPacketPTAAC
ret.ASC, err = ParseASC(md.AFmtBase)
if err != nil {
return ret, err
}
} else { } else {
ret.VideoPayloadType = base.AVPacketPTAVC ret.AudioPayloadType = base.AVPacketPTUnknown
}
if i < len(c.AControlList) {
ret.VideoAControl = c.AControlList[i].Value
}
case base.RTPPacketTypeAAC:
ret.AudioClockRate = item.ClockRate
ret.AudioPayloadType = base.AVPacketPTAAC
if i < len(c.AControlList) {
ret.AudioAControl = c.AControlList[i].Value
} }
default: case "video":
return ret, ErrSDP ret.HasVideo = true
} ret.VideoClockRate = md.ARTPMap.ClockRate
} ret.VideoAControl = md.AControl.Value
for _, item := range c.AFmtPBaseList { ret.VideoPayloadTypeOrigin = md.ARTPMap.PayloadType
switch item.Format { switch md.ARTPMap.EncodingName {
case base.RTPPacketTypeAVCOrHEVC: case ARTPMapEncodingNameH264:
if ret.VideoPayloadType == base.AVPacketPTHEVC { ret.VideoPayloadType = base.AVPacketPTAVC
ret.VPS, ret.SPS, ret.PPS, err = ParseVPSSPSPPS(item) ret.SPS, ret.PPS, err = ParseSPSPPS(md.AFmtBase)
} else { if err != nil {
ret.SPS, ret.PPS, err = ParseSPSPPS(item) return ret, err
} }
if err != nil { case ARTPMapEncodingNameH265:
return ret, err ret.VideoPayloadType = base.AVPacketPTHEVC
} ret.VPS, ret.SPS, ret.PPS, err = ParseVPSSPSPPS(md.AFmtBase)
case base.RTPPacketTypeAAC: if err != nil {
ret.ASC, err = ParseASC(item) return ret, err
if err != nil { }
return ret, err default:
ret.VideoPayloadType = base.AVPacketPTUnknown
} }
default:
return ret, ErrSDP
} }
} }
@ -117,37 +130,74 @@ func ParseSDP2LogicContext(b []byte) (LogicContext, error) {
// 例子见单元测试 // 例子见单元测试
func ParseSDP2RawContext(b []byte) (RawContext, error) { func ParseSDP2RawContext(b []byte) (RawContext, error) {
var sdpCtx RawContext var (
sdpCtx RawContext
md *MediaDesc
)
s := string(b) s := string(b)
lines := strings.Split(s, "\r\n") lines := strings.Split(s, "\r\n")
for _, line := range lines { for _, line := range lines {
if strings.HasPrefix(line, "m=") {
m, err := ParseM(line)
if err != nil {
return sdpCtx, err
}
if md != nil {
sdpCtx.MediaDescList = append(sdpCtx.MediaDescList, *md)
}
md = &MediaDesc{
M: m,
}
}
if strings.HasPrefix(line, "a=rtpmap") { if strings.HasPrefix(line, "a=rtpmap") {
aRTPMap, err := ParseARTPMap(line) aRTPMap, err := ParseARTPMap(line)
if err != nil { if err != nil {
return sdpCtx, err return sdpCtx, err
} }
sdpCtx.ARTPMapList = append(sdpCtx.ARTPMapList, aRTPMap) if md == nil {
continue
}
md.ARTPMap = aRTPMap
} }
if strings.HasPrefix(line, "a=fmtp") { if strings.HasPrefix(line, "a=fmtp") {
aFmtPBase, err := ParseAFmtPBase(line) aFmtPBase, err := ParseAFmtPBase(line)
if err != nil { if err != nil {
return sdpCtx, err return sdpCtx, err
} }
sdpCtx.AFmtPBaseList = append(sdpCtx.AFmtPBaseList, aFmtPBase) if md == nil {
continue
}
md.AFmtBase = aFmtPBase
} }
if strings.HasPrefix(line, "a=control") { if strings.HasPrefix(line, "a=control") {
aControl, err := ParseAControl(line) aControl, err := ParseAControl(line)
if err != nil { if err != nil {
return sdpCtx, err return sdpCtx, err
} }
sdpCtx.AControlList = append(sdpCtx.AControlList, aControl) if md == nil {
continue
}
md.AControl = aControl
} }
} }
if md != nil {
sdpCtx.MediaDescList = append(sdpCtx.MediaDescList, *md)
}
return sdpCtx, nil return sdpCtx, nil
} }
func ParseM(s string) (ret M, err error) {
ss := strings.TrimPrefix(s, "m=")
items := strings.Split(ss, " ")
if len(items) < 1 {
return ret, ErrSDP
}
ret.Media = items[0]
return
}
// 例子见单元测试 // 例子见单元测试
func ParseARTPMap(s string) (ret ARTPMap, err error) { func ParseARTPMap(s string) (ret ARTPMap, err error) {
// rfc 3640 3.3.1. General // rfc 3640 3.3.1. General
@ -262,10 +312,6 @@ func ParseASC(a AFmtPBase) ([]byte, error) {
} }
func ParseVPSSPSPPS(a AFmtPBase) (vps, sps, pps []byte, err error) { func ParseVPSSPSPPS(a AFmtPBase) (vps, sps, pps []byte, err error) {
if a.Format != base.RTPPacketTypeAVCOrHEVC {
return nil, nil, nil, ErrSDP
}
v, ok := a.Parameters["sprop-vps"] v, ok := a.Parameters["sprop-vps"]
if !ok { if !ok {
return nil, nil, nil, ErrSDP return nil, nil, nil, ErrSDP
@ -296,10 +342,6 @@ func ParseVPSSPSPPS(a AFmtPBase) (vps, sps, pps []byte, err error) {
// 解析AVC/H264的sps，pps // 解析AVC/H264的sps，pps
// 例子见单元测试 // 例子见单元测试
func ParseSPSPPS(a AFmtPBase) (sps, pps []byte, err error) { func ParseSPSPPS(a AFmtPBase) (sps, pps []byte, err error) {
if a.Format != base.RTPPacketTypeAVCOrHEVC {
return nil, nil, ErrSDP
}
v, ok := a.Parameters["sprop-parameter-sets"] v, ok := a.Parameters["sprop-parameter-sets"]
if !ok { if !ok {
return nil, nil, ErrSDP return nil, nil, ErrSDP

179
pkg/sdp/sdp_test.go
Unescape Escape View File

@ -10,6 +10,7 @@ package sdp_test
import ( import (
"encoding/hex" "encoding/hex"
"strings"
"testing" "testing"
"github.com/q191201771/lal/pkg/base" "github.com/q191201771/lal/pkg/base"
@ -46,17 +47,6 @@ var goldenPPS = []byte{
0x68, 0xEB, 0xEC, 0xB2, 0x2C, 0x68, 0xEB, 0xEC, 0xB2, 0x2C,
} }
func TestParseSDP2LogicContext(t *testing.T) {
ctx, err := sdp.ParseSDP2LogicContext([]byte(goldenSDP))
assert.Equal(t, nil, err)
assert.Equal(t, 44100, ctx.AudioClockRate)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, base.AVPacketPTAAC, ctx.AudioPayloadType)
assert.Equal(t, base.AVPacketPTAVC, ctx.VideoPayloadType)
assert.Equal(t, "streamid=1", ctx.AudioAControl)
assert.Equal(t, "streamid=0", ctx.VideoAControl)
}
func TestParseSDP2RawContext(t *testing.T) { func TestParseSDP2RawContext(t *testing.T) {
sdpCtx, err := sdp.ParseSDP2RawContext([]byte(goldenSDP)) sdpCtx, err := sdp.ParseSDP2RawContext([]byte(goldenSDP))
assert.Equal(t, nil, err) assert.Equal(t, nil, err)
@ -161,3 +151,170 @@ func TestParseVPSSPSPPS(t *testing.T) {
nazalog.Debugf("%s", hex.Dump(sps)) nazalog.Debugf("%s", hex.Dump(sps))
nazalog.Debugf("%s", hex.Dump(pps)) nazalog.Debugf("%s", hex.Dump(pps))
} }
func TestParseSDP2LogicContext(t *testing.T) {
ctx, err := sdp.ParseSDP2LogicContext([]byte(goldenSDP))
assert.Equal(t, nil, err)
assert.Equal(t, true, ctx.HasAudio)
assert.Equal(t, true, ctx.HasVideo)
assert.Equal(t, 44100, ctx.AudioClockRate)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, 97, ctx.AudioPayloadTypeOrigin)
assert.Equal(t, 96, ctx.VideoPayloadTypeOrigin)
assert.Equal(t, base.AVPacketPTAAC, ctx.AudioPayloadType)
assert.Equal(t, base.AVPacketPTAVC, ctx.VideoPayloadType)
assert.Equal(t, "streamid=1", ctx.AudioAControl)
assert.Equal(t, "streamid=0", ctx.VideoAControl)
assert.IsNotNil(t, ctx.ASC)
assert.Equal(t, nil, ctx.VPS)
assert.IsNotNil(t, ctx.SPS)
assert.IsNotNil(t, ctx.PPS)
}
func TestCase2(t *testing.T) {
golden := `v=0
o=- 2252316233 2252316233 IN IP4 0.0.0.0
s=Media Server
c=IN IP4 0.0.0.0
t=0 0
a=control:*
a=packetization-supported:DH
a=rtppayload-supported:DH
a=range:npt=now-
m=video 0 RTP/AVP 98
a=control:trackID=0
a=framerate:25.000000
a=rtpmap:98 H265/90000
a=fmtp:98 profile-id=1;sprop-sps=QgEBAWAAAAMAsAAAAwAAAwBaoAWCAJBY2uSTL5A=;sprop-pps=RAHA8vA8kA==;sprop-vps=QAEMAf//AWAAAAMAsAAAAwAAAwBarAk=
a=recvonly`
golden = strings.ReplaceAll(golden, "\n", "\r\n")
ctx, err := sdp.ParseSDP2LogicContext([]byte(golden))
assert.Equal(t, nil, err)
assert.Equal(t, false, ctx.HasAudio)
assert.Equal(t, true, ctx.HasVideo)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, 98, ctx.VideoPayloadTypeOrigin)
assert.Equal(t, base.AVPacketPTHEVC, ctx.VideoPayloadType)
assert.Equal(t, "trackID=0", ctx.VideoAControl)
assert.Equal(t, nil, ctx.ASC)
assert.IsNotNil(t, ctx.VPS)
assert.IsNotNil(t, ctx.SPS)
assert.IsNotNil(t, ctx.PPS)
nazalog.Debugf("%+v", ctx)
}
func TestCase3(t *testing.T) {
golden := `v=0
o=- 2252310609 2252310609 IN IP4 0.0.0.0
s=Media Server
c=IN IP4 0.0.0.0
t=0 0
a=control:*
a=packetization-supported:DH
a=rtppayload-supported:DH
a=range:npt=now-
m=video 0 RTP/AVP 96
a=control:trackID=0
a=framerate:25.000000
a=rtpmap:96 H264/90000
a=fmtp:96 packetization-mode=1;profile-level-id=4D002A;sprop-parameter-sets=Z00AKp2oHgCJ+WbgICAoAAAfQAAGGoQgAA==,aO48gAA=
a=recvonly
m=audio 0 RTP/AVP 97
a=control:trackID=1
a=rtpmap:97 MPEG4-GENERIC/48000
a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1188
a=recvonly`
golden = strings.ReplaceAll(golden, "\n", "\r\n")
ctx, err := sdp.ParseSDP2LogicContext([]byte(golden))
assert.Equal(t, nil, err)
assert.Equal(t, true, ctx.HasAudio)
assert.Equal(t, true, ctx.HasVideo)
assert.Equal(t, 48000, ctx.AudioClockRate)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, 97, ctx.AudioPayloadTypeOrigin)
assert.Equal(t, 96, ctx.VideoPayloadTypeOrigin)
assert.Equal(t, base.AVPacketPTAAC, ctx.AudioPayloadType)
assert.Equal(t, base.AVPacketPTAVC, ctx.VideoPayloadType)
assert.Equal(t, "trackID=1", ctx.AudioAControl)
assert.Equal(t, "trackID=0", ctx.VideoAControl)
assert.IsNotNil(t, ctx.ASC)
assert.Equal(t, nil, ctx.VPS)
assert.IsNotNil(t, ctx.SPS)
assert.IsNotNil(t, ctx.PPS)
nazalog.Debugf("%+v", ctx)
}
func TestCase4(t *testing.T) {
golden := `v=0
o=- 1109162014219182 0 IN IP4 0.0.0.0
s=HIK Media Server V3.4.103
i=HIK Media Server Session Description : standard
e=NONE
c=IN IP4 0.0.0.0
t=0 0
a=control:*
b=AS:1034
a=range:npt=now-
m=video 0 RTP/AVP 96
i=Video Media
a=rtpmap:96 H264/90000
a=fmtp:96 profile-level-id=4D0014;packetization-mode=0;sprop-parameter-sets=Z2QAIK2EAQwgCGEAQwgCGEAQwgCEO1AoA803AQEBQAAAAwBAAAAMoQ==,aO48sA==
a=control:trackID=video
b=AS:1024
m=audio 0 RTP/AVP 8
i=Audio Media
a=rtpmap:8 PCMA/8000
a=control:trackID=audio
b=AS:10
a=Media_header:MEDIAINFO=494D4B48020100000400000111710110401F000000FA000000000000000000000000000000000000;
a=appversion:1.0`
golden = strings.ReplaceAll(golden, "\n", "\r\n")
ctx, err := sdp.ParseSDP2LogicContext([]byte(golden))
assert.Equal(t, nil, err)
assert.Equal(t, true, ctx.HasAudio)
assert.Equal(t, true, ctx.HasVideo)
assert.Equal(t, 8000, ctx.AudioClockRate)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, 8, ctx.AudioPayloadTypeOrigin)
assert.Equal(t, 96, ctx.VideoPayloadTypeOrigin)
//assert.Equal(t, base.AVPacketPTAAC, ctx.AudioPayloadType)
assert.Equal(t, base.AVPacketPTAVC, ctx.VideoPayloadType)
assert.Equal(t, "trackID=audio", ctx.AudioAControl)
assert.Equal(t, "trackID=video", ctx.VideoAControl)
assert.Equal(t, nil, ctx.ASC)
assert.Equal(t, nil, ctx.VPS)
assert.IsNotNil(t, ctx.SPS)
assert.IsNotNil(t, ctx.PPS)
nazalog.Debugf("%+v", ctx)
}
func TestCase5(t *testing.T) {
golden := `v=0
o=- 1001 1 IN IP4 192.168.0.221
s=VCP IPC Realtime stream
m=video 0 RTP/AVP 105
c=IN IP4 192.168.0.221
a=control:rtsp://192.168.0.221/media/video1/video
a=rtpmap:105 H264/90000
a=fmtp:105 profile-level-id=64002a; packetization-mode=1; sprop-parameter-sets=Z2QAKq2EAQwgCGEAQwgCGEAQwgCEO1A8ARPyzcBAQFAAAD6AAAnECEA=,aO4xshs=
a=recvonly
m=application 0 RTP/AVP 107
c=IN IP4 192.168.0.221
a=control:rtsp://192.168.0.221/media/video1/metadata
a=rtpmap:107 vnd.onvif.metadata/90000
a=fmtp:107 DecoderTag=h3c-v3 RTCP=0
a=recvonly`
golden = strings.ReplaceAll(golden, "\n", "\r\n")
ctx, err := sdp.ParseSDP2LogicContext([]byte(golden))
assert.Equal(t, nil, err)
assert.Equal(t, false, ctx.HasAudio)
assert.Equal(t, true, ctx.HasVideo)
assert.Equal(t, 90000, ctx.VideoClockRate)
assert.Equal(t, 105, ctx.VideoPayloadTypeOrigin)
assert.Equal(t, base.AVPacketPTAVC, ctx.VideoPayloadType)
assert.Equal(t, "rtsp://192.168.0.221/media/video1/video", ctx.VideoAControl)
assert.Equal(t, nil, ctx.VPS)
assert.IsNotNil(t, ctx.SPS)
assert.IsNotNil(t, ctx.PPS)
nazalog.Debugf("%+v", ctx)
}

Write Preview
Loading…
Cancel
Save