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

[fix] package rtmp: 接收rtmp数据时，同一个message的多个chunk混合使用fmt1，2时，可能出现时间戳多加的情况

Browse Source
pull/2/head
q191201771
parent 9e80735f1a
commit 99ab8df79a
3 changed files with 33 additions and 8 deletions

35
pkg/rtmp/chunk_composer.go
Unescape Escape View File

@ -40,16 +40,19 @@ type CompleteMessageCB func(stream *Stream) error
// @param cb 回调结束后，内存块会被 ChunkComposer 再次使用
func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
bootstrap := make([]byte, 11)
absTsFlag := false
for {
// 5.3.1.1. Chunk Basic Header
// 读取fmt和csid
if _, err := io.ReadAtLeast(reader, bootstrap[:1], 1); err != nil {
return err
}
// 5.3.1.1. Chunk Basic Header
fmt := (bootstrap[0] >> 6) & 0x03
csid := int(bootstrap[0] & 0x3f)
// csid可能是变长的
switch csid {
case 0:
if _, err := io.ReadAtLeast(reader, bootstrap[:1], 1); err != nil {
@ -68,6 +71,7 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
stream := c.getOrCreateStream(csid)
// 5.3.1.2. Chunk Message Header
// 当前chunk的fmt不同，Message Header包含的字段也不同，是变长
switch fmt {
case 0:
if _, err := io.ReadAtLeast(reader, bootstrap[:11], 11); err != nil {
@ -76,6 +80,7 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
// 包头中为绝对时间戳
stream.header.Timestamp = bele.BEUint24(bootstrap)
stream.header.TimestampAbs = stream.header.Timestamp
absTsFlag = true
stream.header.MsgLen = bele.BEUint24(bootstrap[3:])
stream.header.MsgTypeID = bootstrap[6]
stream.header.MsgStreamID = int(bele.LEUint32(bootstrap[7:]))
@ -87,7 +92,7 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
}
// 包头中为相对时间戳
stream.header.Timestamp = bele.BEUint24(bootstrap)
stream.header.TimestampAbs += stream.header.Timestamp
//stream.header.TimestampAbs += stream.header.Timestamp
stream.header.MsgLen = bele.BEUint24(bootstrap[3:])
stream.header.MsgTypeID = bootstrap[6]
@ -98,7 +103,7 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
}
// 包头中为相对时间戳
stream.header.Timestamp = bele.BEUint24(bootstrap)
stream.header.TimestampAbs += stream.header.Timestamp
//stream.header.TimestampAbs += stream.header.Timestamp
case 3:
// noop
@ -107,7 +112,9 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
// 5.3.1.3 Extended Timestamp
// 使用ffmpeg推流时，发现时间戳超过3字节最大值后，即使是fmt3(即包头大小为0)，依然存在ext ts字段
// 所以这里我将 `==` 的判断改成了 `>=`
// TODO chef: 测试其他客户端和ext ts相关的表现
// TODO chef:
// - 测试其他客户端和ext ts相关的表现
// - 这部分可能还有问题，需要根据具体的case调整
//if stream.header.Timestamp == maxTimestampInMessageHeader {
if stream.header.Timestamp >= maxTimestampInMessageHeader {
if _, err := io.ReadAtLeast(reader, bootstrap[:4], 4); err != nil {
@ -126,7 +133,7 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
}
}
//stream.header.CSID = csid
//log.Debugf("CHEFGREPME tag1 fmt:%d header:%+v csid:%d len:%d ts:%d", fmt, stream.header, csid, stream.header.MsgLen, stream.header.TimestampAbs)
//nazalog.Debugf("ChunkComposer chunk fmt:%d header:%+v csid:%d len:%d ts:%d", fmt, stream.header, csid, stream.header.MsgLen, stream.header.TimestampAbs)
var neededSize uint32
if stream.header.MsgLen <= c.peerChunkSize {
@ -152,6 +159,13 @@ func (c *ChunkComposer) RunLoop(reader io.Reader, cb CompleteMessageCB) error {
}
stream.header.CSID = csid
if !absTsFlag {
// 这么处理相当于取最后一个chunk的时间戳差值，有的协议栈是取的第一个，正常来说都可以
stream.header.TimestampAbs += stream.header.Timestamp
}
absTsFlag = false
//nazalog.Debugf("ChunkComposer message fmt:%d header:%+v csid:%d len:%d ts:%d", fmt, stream.header, csid, stream.header.MsgLen, stream.header.TimestampAbs)
if err := cb(stream); err != nil {
return err
}
@ -171,3 +185,14 @@ func (c *ChunkComposer) getOrCreateStream(csid int) *Stream {
}
return stream
}
// 临时存放一些rtmp推流case在这，便于理解，以及修改后，回归用
//
// 场景：ffmpeg推送test.flv至lals
// 关注点：message超过chunk时，fmt和timestamp的值
//
// ChunkComposer chunk fmt:1 header:{CSID:6 MsgLen:143 Timestamp:40 MsgTypeID:9 MsgStreamID:1 TimestampAbs:520} csid:6 len:143 ts:520
// ChunkComposer chunk fmt:1 header:{CSID:6 MsgLen:4511 Timestamp:40 MsgTypeID:9 MsgStreamID:1 TimestampAbs:560} csid:6 len:4511 ts:560
// ChunkComposer chunk fmt:3 header:{CSID:6 MsgLen:4511 Timestamp:40 MsgTypeID:9 MsgStreamID:1 TimestampAbs:560} csid:6 len:4511 ts:560
// 此处应只给上层返回一次，也即一个message，时间戳应该是560
// ChunkComposer chunk fmt:1 header:{CSID:6 MsgLen:904 Timestamp:40 MsgTypeID:9 MsgStreamID:1 TimestampAbs:600} csid:6 len:904 ts:600

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

@ -193,7 +193,7 @@ func (s *HandshakeServer) ReadC0C1(reader io.Reader) (err error) {
// s2
// make digest to s2 suffix position
random1528(s2)
replyOffs := s2Len - keyLen
makeDigestWithoutCenterPart(s2, replyOffs, s2key, s2[replyOffs:])
}
@ -287,4 +287,3 @@ func init() {
random1528Buf[i] = bs[i%bsl]
}
}

3
pkg/rtmp/stream.go
Unescape Escape View File

@ -17,10 +17,11 @@ import (
const initMsgLen = 4096
// TODO chef: 将Timestamp字段隐藏，不对外暴露
type Header struct {
CSID int
MsgLen uint32 // 不包含header的大小
Timestamp uint32 // NOTICE 是 rtmp 协议 header 中的时间戳，可能是绝对的，也可能是相对的。
Timestamp uint32 // NOTICE 是 rtmp 协议 header 中的时间戳，可能是绝对的，也可能是相对的。上层不应该使用这个字段，而应该使用TimestampAbs
MsgTypeID uint8 // 8 audio 9 video 18 metadata
MsgStreamID int

Write Preview
Loading…
Cancel
Save