mirror of https://github.com/q191201771/lal.git
1. [feat] 新增app/demo/calcrtmpdelay，可用于测量rtmp服务器的转发延时 2. [refactor] 重构app/demo/pullrtmp2pushrtmp的对象管理逻辑
q191201771
4 years ago
parent
fd0f71f9af
commit
4a48518f5a
@ -0,0 +1,134 @@
|
||||
// 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 main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/q191201771/lal/pkg/base"
|
||||
"github.com/q191201771/lal/pkg/httpflv"
|
||||
"github.com/q191201771/lal/pkg/remux"
|
||||
"github.com/q191201771/lal/pkg/rtmp"
|
||||
"github.com/q191201771/naza/pkg/nazalog"
|
||||
"github.com/q191201771/naza/pkg/nazamd5"
|
||||
)
|
||||
|
||||
const detailFilename = "delay.txt"
|
||||
|
||||
func main() {
|
||||
tagKey2writeTime := make(map[string]time.Time)
|
||||
var delays []int64
|
||||
var mu sync.Mutex
|
||||
|
||||
_ = nazalog.Init(func(option *nazalog.Option) {
|
||||
option.AssertBehavior = nazalog.AssertFatal
|
||||
})
|
||||
filename, pushURL, pullURL := parseFlag()
|
||||
|
||||
tags, err := httpflv.ReadAllTagsFromFLVFile(filename)
|
||||
nazalog.Assert(nil, err)
|
||||
nazalog.Infof("read tags from flv file succ. len of tags=%d", len(tags))
|
||||
|
||||
pushSession := rtmp.NewPushSession()
|
||||
err = pushSession.Push(pushURL)
|
||||
nazalog.Assert(nil, err)
|
||||
nazalog.Info("push succ.")
|
||||
//defer pushSession.Dispose()
|
||||
|
||||
pullSession := rtmp.NewPullSession()
|
||||
err = pullSession.Pull(pullURL, func(msg base.RTMPMsg) {
|
||||
tagKey := nazamd5.MD5(msg.Payload)
|
||||
mu.Lock()
|
||||
t, exist := tagKey2writeTime[tagKey]
|
||||
if !exist {
|
||||
nazalog.Errorf("tag key not exist.")
|
||||
} else {
|
||||
delay := time.Now().Sub(t).Milliseconds()
|
||||
delays = append(delays, delay)
|
||||
delete(tagKey2writeTime, tagKey)
|
||||
}
|
||||
mu.Unlock()
|
||||
})
|
||||
nazalog.Assert(nil, err)
|
||||
nazalog.Info("pull succ.")
|
||||
//defer pullSession.Dispose()
|
||||
|
||||
go func() {
|
||||
for {
|
||||
time.Sleep(5 * time.Second)
|
||||
pushSession.UpdateStat(1)
|
||||
pullSession.UpdateStat(1)
|
||||
nazalog.Debugf("stat bitrate. push=%+v, pull=%+v", pushSession.GetStat().Bitrate, pullSession.GetStat().Bitrate)
|
||||
}
|
||||
}()
|
||||
|
||||
prevTS := int64(-1)
|
||||
for _, tag := range tags {
|
||||
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 {
|
||||
nazalog.Errorf("tag key already exist. key=%s", tagKey)
|
||||
}
|
||||
tagKey2writeTime[tagKey] = time.Now()
|
||||
mu.Unlock()
|
||||
|
||||
err = pushSession.AsyncWrite(chunks)
|
||||
nazalog.Assert(nil, err)
|
||||
//nazalog.Debugf("sent. %d", i)
|
||||
}
|
||||
|
||||
min := int64(2147483647)
|
||||
max := int64(0)
|
||||
avg := int64(0)
|
||||
sum := int64(0)
|
||||
fp, _ := os.Create(detailFilename)
|
||||
defer fp.Close()
|
||||
for _, d := range delays {
|
||||
if d < min {
|
||||
min = d
|
||||
}
|
||||
if d > max {
|
||||
max = d
|
||||
}
|
||||
sum += d
|
||||
_, _ = fp.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)
|
||||
}
|
||||
|
||||
func parseFlag() (filename, pushURL, pullURL string) {
|
||||
f := flag.String("f", "", "specify flv file")
|
||||
i := flag.String("i", "", "specify rtmp pull url")
|
||||
o := flag.String("o", "", "specify rtmp push url")
|
||||
flag.Parse()
|
||||
if *f == "" || *i == "" || *o == "" {
|
||||
flag.Usage()
|
||||
_, _ = fmt.Fprintf(os.Stderr, `Example:
|
||||
%s -f test.flv -i rtmp://127.0.0.1:1935/live/test -o rtmp://127.0.0.1:1935/live/test
|
||||
`, os.Args[0])
|
||||
base.OSExitAndWaitPressIfWindows(1)
|
||||
}
|
||||
return *f, *i, *o
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
{
|
||||
"rtmp": {
|
||||
"enable": true,
|
||||
"addr": ":1935",
|
||||
"gop_num": 0
|
||||
},
|
||||
"httpflv": {
|
||||
"enable": false,
|
||||
"sub_listen_addr": ":8080",
|
||||
"enable_https": false,
|
||||
"https_addr": ":4433",
|
||||
"https_cert_file": "./conf/cert.pem",
|
||||
"https_key_file": "./conf/key.pem",
|
||||
"gop_num": 0
|
||||
},
|
||||
"hls": {
|
||||
"enable": false,
|
||||
"sub_listen_addr": ":8081",
|
||||
"out_path": "/tmp/lal/hls/",
|
||||
"fragment_duration_ms": 3000,
|
||||
"fragment_num": 6,
|
||||
"cleanup_flag": true
|
||||
},
|
||||
"httpts": {
|
||||
"enable": false,
|
||||
"sub_listen_addr": ":8082"
|
||||
},
|
||||
"rtsp": {
|
||||
"enable": false,
|
||||
"addr": ":5544"
|
||||
},
|
||||
"relay_push": {
|
||||
"enable": false,
|
||||
"addr_list":[
|
||||
]
|
||||
},
|
||||
"relay_pull": {
|
||||
"enable": false,
|
||||
"addr": ""
|
||||
},
|
||||
"http_api": {
|
||||
"enable": false,
|
||||
"addr": ":8083"
|
||||
},
|
||||
"server_id": "1",
|
||||
"http_notify": {
|
||||
"enable": false,
|
||||
"update_interval_sec": 5,
|
||||
"on_server_start": "http://127.0.0.1:10101/on_server_start",
|
||||
"on_update": "http://127.0.0.1:10101/on_update",
|
||||
"on_pub_start": "http://127.0.0.1:10101/on_pub_start",
|
||||
"on_pub_stop": "http://127.0.0.1:10101/on_pub_stop",
|
||||
"on_sub_start": "http://127.0.0.1:10101/on_sub_start",
|
||||
"on_sub_stop": "http://127.0.0.1:10101/on_sub_stop",
|
||||
"on_rtmp_connect": "http://127.0.0.1:10101/on_rtmp_connect"
|
||||
},
|
||||
"pprof": {
|
||||
"enable": true,
|
||||
"addr": ":8084"
|
||||
},
|
||||
"log": {
|
||||
"level": 1,
|
||||
"filename": "./logs/lalserver.log",
|
||||
"is_to_stdout": true,
|
||||
"is_rotate_daily": true,
|
||||
"short_file_flag": true,
|
||||
"assert_behavior": 1
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,34 @@
|
||||
// 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 "io"
|
||||
|
||||
func ReadAllTagsFromFLVFile(filename string) ([]Tag, error) {
|
||||
var tags []Tag
|
||||
|
||||
var ffr FLVFileReader
|
||||
err := ffr.Open(filename)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for {
|
||||
tag, err := ffr.ReadTag()
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
return tags, nil
|
||||
} else {
|
||||
return tags, err
|
||||
}
|
||||
}
|
||||
tags = append(tags, tag)
|
||||
}
|
||||
// never reach here
|
||||
}
|
||||
Loading…
Reference in New Issue