[refator] package httpflv: 新增FlvFilePump，可循环匀速读取flv文件

app/demo/calcrtmpdelay/calcrtmpdelay.go

@@ -24,7 +24,7 @@ import (
 	"github.com/q191201771/naza/pkg/nazamd5"
 )
 
-const detailFilename = "delay.txt"
+const outDetailFilename = "delay.txt"
 
 type PullType int
 
@@ -54,24 +54,18 @@ func main() {
 	})
 	defer nazalog.Sync()
 
+	var mu sync.Mutex
 	tagKey2writeTime := make(map[string]time.Time)
 	var delays []int64
-	var mu sync.Mutex
 
 	filename, pushUrl, pullUrl, pullType := parseFlag()
 	nazalog.Infof("parse flag succ. filename=%s, pushUrl=%s, pullUrl=%s, pullType=%s",
 		filename, pushUrl, pullUrl, pullType.Readable())
 
-	tags, err := httpflv.ReadAllTagsFromFlvFile(filename)
-	if err != nil {
-		nazalog.Fatalf("read tags from flv file failed. err=%+v", err)
-	}
-	nazalog.Infof("read tags from flv file succ. len of tags=%d", len(tags))
-
 	pushSession := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
 		option.PushTimeoutMs = 10000
 	})
-	err = pushSession.Push(pushUrl)
+	err := pushSession.Push(pushUrl)
 	if err != nil {
 		nazalog.Fatalf("push rtmp failed. url=%s, err=%+v", pushUrl, err)
 	}
@@ -135,17 +129,14 @@ func main() {
 		}
 	}()
 
-	prevTs := int64(-1)
-	for _, tag := range tags {
+	// 读取flv文件
+	flvFilePump := httpflv.NewFileFilePump(func(option *httpflv.FlvFilePumpOption) {
+		option.IsRecursive = false
+	})
+	err = flvFilePump.Pump(filename, func(tag httpflv.Tag) bool {
 		h := remux.FlvTagHeader2RtmpHeader(tag.Header)
 		chunks := rtmp.Message2Chunks(tag.Raw[11:11+h.MsgLen], &h)
 
-		if prevTs >= 0 && int64(h.TimestampAbs) > prevTs {
-			diff := int64(h.TimestampAbs) - prevTs
-			time.Sleep(time.Duration(diff) * time.Millisecond)
-		}
-		prevTs = int64(h.TimestampAbs)
-
 		mu.Lock()
 		tagKey := nazamd5.Md5(tag.Raw[11 : 11+h.MsgLen])
 		if _, exist := tagKey2writeTime[tagKey]; exist {
@@ -157,18 +148,24 @@ func main() {
 		err = pushSession.Write(chunks)
 		if err != nil {
 			nazalog.Fatalf("write failed. err=%+v", err)
+			return false
 		}
-		//nazalog.Debugf("sent. %d", i)
+		return true
+	})
+	if err != nil {
+		nazalog.Fatalf("pump flv file failed. err=%+v", err)
 	}
+
 	_ = pushSession.Flush()
 	time.Sleep(300 * time.Millisecond)
 
+	// 信息分析汇总输出
 	min := int64(2147483647)
 	max := int64(0)
 	avg := int64(0)
 	sum := int64(0)
-	fp, _ := os.Create(detailFilename)
-	defer fp.Close()
+	outDetailFp, _ := os.Create(outDetailFilename)
+	defer outDetailFp.Close()
 	for _, d := range delays {
 		if d < min {
 			min = d
@@ -177,12 +174,12 @@ func main() {
 			max = d
 		}
 		sum += d
-		_, _ = fp.WriteString(fmt.Sprintf("%d\n", d))
+		_, _ = outDetailFp.WriteString(fmt.Sprintf("%d\n", d))
 	}
 	if len(delays) > 0 {
 		avg = sum / int64(len(delays))
 	}
-	nazalog.Debugf("len(tagKey2writeTime)=%d, delays(len=%d, avg=%d, min=%d, max=%d), detailFilename=%s", len(tagKey2writeTime), len(delays), avg, min, max, detailFilename)
+	nazalog.Debugf("len(tagKey2writeTime)=%d, delays(len=%d, avg=%d, min=%d, max=%d), detailFilename=%s", len(tagKey2writeTime), len(delays), avg, min, max, outDetailFilename)
 }
 
 func parseFlag() (filename, pushUrl, pullUrl string, pullType PullType) {

app/demo/pushrtmp/pushrtmp.go

@@ -59,8 +59,9 @@ func main() {
 	urls := collect(urlTmpl, num)
 
 	tags, err := httpflv.ReadAllTagsFromFlvFile(filename)
-	if err != nil {
-		nazalog.Fatalf("read tags from flv file failed. err=%+v", err)
+	if err != nil || len(tags) == 0 {
+		nazalog.Fatalf("read tags from flv file failed. len=%d, err=%+v", len(tags), err)
+		return
 	}
 	nazalog.Infof("read tags from flv file succ. len of tags=%d", len(tags))
 
@@ -75,7 +76,7 @@ func main() {
 	wg.Add(len(urls))
 	for _, url := range urls {
 		go func(u string) {
-			push(tags, []string{u}, isRecursive)
+			push(tags, u, isRecursive)
 			wg.Done()
 			atomic.AddInt32(&aliveSessionCount, -1)
 		}(url)
@@ -85,141 +86,33 @@ func main() {
 	nazalog.Info("< main.")
 }
 
-func push(tags []httpflv.Tag, urls []string, isRecursive bool) {
-	var sessionList []*rtmp.PushSession
+func push(tags []httpflv.Tag, url string, isRecursive bool) {
+	ps := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
+		option.PushTimeoutMs = 5000
+		option.WriteAvTimeoutMs = 10000
+	})
 
-	if len(tags) == 0 || len(urls) == 0 {
+	if err := ps.Push(url); err != nil {
+		nazalog.Errorf("push failed. err=%v", err)
 		return
 	}
+	atomic.AddInt32(&aliveSessionCount, 1)
 
-	var err error
-
-	for i := range urls {
-		ps := rtmp.NewPushSession(func(option *rtmp.PushSessionOption) {
-			option.PushTimeoutMs = 5000
-			option.WriteAvTimeoutMs = 10000
-		})
-
-		err = ps.Push(urls[i])
-		if err != nil {
-			nazalog.Errorf("push failed. err=%v", err)
-			continue
-		}
-		atomic.AddInt32(&aliveSessionCount, 1)
-
-		nazalog.Infof("push succ. url=%s", urls[i])
-		sessionList = append(sessionList, ps)
-	}
-	check(sessionList)
-
-	var totalBaseTs uint32 // 每轮最后更新
-	var prevTs uint32      // 上一个tag
-	var hasReadThisBaseTs bool
-	var thisBaseTs uint32 // 每轮第一个tag
-	var hasTraceFirstTagTs bool
-	var firstTagTs uint32  // 所有轮第一个tag
-	var firstTagTick int64 // 所有轮第一个tag的物理发送时间
-
-	// 1. 保证metadata只在最初发送一次
-	// 2. 多轮，时间戳会翻转，需要处理，让它线性增长
-
-	// 多轮，一个循环代表一次完整文件的发送
-	for i := 0; ; i++ {
-		nazalog.Infof(" > round. i=%d, totalBaseTs=%d, prevTs=%d, thisBaseTs=%d",
-			i, totalBaseTs, prevTs, thisBaseTs)
-
-		hasReadThisBaseTs = false
-
-		// 一轮，遍历文件的所有tag数据
-		for _, tag := range tags {
-			h := remux.FlvTagHeader2RtmpHeader(tag.Header)
-
-			// metadata只发送一次
-			if tag.IsMetadata() {
-				if totalBaseTs == 0 {
-					h.TimestampAbs = 0
-					chunks := rtmp.Message2Chunks(tag.Raw[11:11+h.MsgLen], &h)
-					send(sessionList, chunks)
-				} else {
-					// noop
-				}
-				continue
-			}
-
-			if hasReadThisBaseTs {
-				// 本轮非第一个tag
-
-				// 之前已经读到了这轮读文件的base值，ts要减去base
-				h.TimestampAbs = tag.Header.Timestamp - thisBaseTs + totalBaseTs
-			} else {
-				// 本轮第一个tag
-
-				// 设置base，ts设置为上一轮读文件的值
-				thisBaseTs = tag.Header.Timestamp
-				h.TimestampAbs = totalBaseTs
-				hasReadThisBaseTs = true
-			}
+	nazalog.Infof("push succ. url=%s", url)
 
-			if h.TimestampAbs < prevTs {
-				// ts比上一个包的还小，直接设置为上一包的值，并且不sleep直接发送
-				h.TimestampAbs = prevTs
-				nazalog.Errorf("this tag timestamp less than prev timestamp. h.TimestampAbs=%d, prevTs=%d", h.TimestampAbs, prevTs)
-			}
+	flvFilePump := httpflv.NewFileFilePump(func(option *httpflv.FlvFilePumpOption) {
+		option.IsRecursive = isRecursive
+	})
+	_ = flvFilePump.PumpWithTags(tags, func(tag httpflv.Tag) bool {
+		h := remux.FlvTagHeader2RtmpHeader(tag.Header)
+		chunks := rtmp.Message2Chunks(tag.Raw[11:11+h.MsgLen], &h)
 
-			chunks := rtmp.Message2Chunks(tag.Raw[11:11+h.MsgLen], &h)
-
-			if hasTraceFirstTagTs {
-				// 所有轮的非第一个tag
-
-				// 当前距离第一个tag的物理发送时间，以及距离第一个tag的时间戳
-				// 如果物理时间短，就睡眠相应的时间
-				n := time.Now().UnixNano() / 1000000
-				diffTick := n - firstTagTick
-				diffTs := h.TimestampAbs - firstTagTs
-				if diffTick < int64(diffTs) {
-					time.Sleep(time.Duration(int64(diffTs)-diffTick) * time.Millisecond)
-				}
-			} else {
-				// 所有轮的第一个tag
-
-				// 记录所有轮的第一个tag的物理发送时间，以及数据的时间戳
-				firstTagTick = time.Now().UnixNano() / 1000000
-				firstTagTs = h.TimestampAbs
-				hasTraceFirstTagTs = true
-			}
-
-			send(sessionList, chunks)
-
-			prevTs = h.TimestampAbs
-		} // tags for loop
-
-		totalBaseTs = prevTs + 1
-
-		if !isRecursive {
-			break
-		}
-	}
-}
-
-func send(sessionList []*rtmp.PushSession, b []byte) {
-	var s []*rtmp.PushSession
-	for _, ps := range sessionList {
-		if err := ps.Write(b); err != nil {
+		if err := ps.Write(chunks); err != nil {
 			nazalog.Errorf("write data error. err=%v", err)
-			continue
+			return false
 		}
-		s = append(s, ps)
-	}
-	sessionList = s
-
-	check(sessionList)
-}
-
-func check(sessionList []*rtmp.PushSession) {
-	if len(sessionList) == 0 {
-		nazalog.Errorf("all push session dead.")
-		os.Exit(1)
-	}
+		return true
+	})
 }
 
 func collect(urlTmpl string, num int) (urls []string) {

pkg/httpflv/flv_file_pump.go

@@ -0,0 +1,143 @@
+// Copyright 2021, 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 httpflv
+
+import (
+	"time"
+
+	"github.com/q191201771/naza/pkg/nazalog"
+)
+
+// 读取flv文件，将tag按时间戳间隔缓慢（类似于ffmpeg的-re）返回
+
+type FlvFilePumpOption struct {
+	IsRecursive bool // 如果为true，则循环返回文件内容（类似于ffmpeg的-stream_loop -1）
+}
+
+var defaultFlvFilePumpOption = FlvFilePumpOption{
+	IsRecursive: false,
+}
+
+type FlvFilePump struct {
+	option FlvFilePumpOption
+}
+
+type ModFlvFilePumpOption func(option *FlvFilePumpOption)
+
+func NewFileFilePump(modOptions ...ModFlvFilePumpOption) *FlvFilePump {
+	option := defaultFlvFilePumpOption
+	for _, fn := range modOptions {
+		fn(&option)
+	}
+
+	return &FlvFilePump{option: option}
+}
+
+type OnPumpFlvTag func(tag Tag) bool
+
+// @param onFlvTag 如果回调中返回false，则停止Pump
+//
+func (f *FlvFilePump) Pump(filename string, onFlvTag OnPumpFlvTag) error {
+	// 一次性将文件所有内容读入内存，后续不再读取文件
+	tags, err := ReadAllTagsFromFlvFile(filename)
+	if err != nil {
+		return err
+	}
+
+	return f.PumpWithTags(tags, onFlvTag)
+}
+
+// @return error 暂时只做预留，目前只会返回nil
+//
+func (f *FlvFilePump) PumpWithTags(tags []Tag, onFlvTag OnPumpFlvTag) error {
+	var totalBaseTs uint32 // 整体的基础时间戳。每轮最后更新
+
+	var hasReadThisBaseTs bool
+	var thisBaseTs uint32 // 每一轮的第一个tag时间戳
+
+	var prevTagTs uint32 // 上一个tag的时间戳
+
+	var hasReadTotalFirstTag bool
+	var totalFirstTagTs uint32  // 第一轮的第一个tag的时间戳
+	var totalFirstTagTick int64 // 第一轮的第一个tag的物理时间
+
+	const addTsBetweenRound = 1
+
+	// 循环一次，代表遍历文件一次
+	for roundIndex := 0; ; roundIndex++ {
+		nazalog.Debugf("new round. index=%d", roundIndex)
+
+		hasReadThisBaseTs = false
+
+		// 遍历所有tag数据
+		for _, tag := range tags {
+			// metadata只在第一轮发送一次
+			if tag.IsMetadata() {
+				if totalBaseTs == 0 {
+					tag.Header.Timestamp = 0
+					if !onFlvTag(tag) {
+						return nil
+					}
+				}
+				continue
+			}
+
+			// 修改时间戳
+			// 使得不同轮依然线性增长
+			if !hasReadThisBaseTs {
+				// 本轮第一个tag
+
+				thisBaseTs = tag.Header.Timestamp
+				hasReadThisBaseTs = true
+
+				tag.Header.Timestamp = totalBaseTs
+			} else {
+				tag.Header.Timestamp = totalBaseTs + tag.Header.Timestamp - thisBaseTs
+			}
+
+			// 修改时间戳
+			// 如果时间戳比前一个tag的还小，可能发生了跳跃，我们直接设置为上一包的值+1，然后不sleep直接发送
+			if tag.Header.Timestamp < prevTagTs {
+				tag.Header.Timestamp = prevTagTs + 1
+			}
+
+			if hasReadTotalFirstTag {
+				// 当前时间戳与第一轮的第一个tag的时间戳差值
+				diffTs := tag.Header.Timestamp - totalFirstTagTs
+
+				// 当前物理时间与第一轮的第一个tag的物理时间差值
+				diffTick := time.Now().UnixNano()/1000000 - totalFirstTagTick
+
+				// 如果还没到物理时间差值，就sleep
+				if diffTick < int64(diffTs) {
+					time.Sleep(time.Duration(int64(diffTs)-diffTick) * time.Millisecond)
+				}
+			} else {
+				// 第一轮的第一个tag，记录下来
+
+				totalFirstTagTick = time.Now().UnixNano() / 1000000
+				totalFirstTagTs = tag.Header.Timestamp
+				hasReadTotalFirstTag = true
+			}
+
+			if !onFlvTag(tag) {
+				return nil
+			}
+
+			prevTagTs = tag.Header.Timestamp
+		}
+
+		totalBaseTs = prevTagTs + addTsBetweenRound
+
+		if !f.option.IsRecursive {
+			break
+		}
+	}
+	return nil
+}