messages:

- #77 [feat] lalserver: httpflv,httpts,hls的url pattern路由路径可以在配置文件中配置
- #64 [feat] hls默认提供两种播放url地址
- #77 [refactor] package hls: 将http url路径格式,文件存储路径格式,文件命名格式,映射关系抽象出来,业务方可在外层实现IPathSolver接口做定制
pull/95/head
q191201771 4 years ago
parent 229b55936d
commit 0cf8660d17

3
.gitignore vendored

@ -1,5 +1,3 @@
/app/demo/pubrtmp2pushrtmp
profile.out
coverage.html
*.aac
@ -8,7 +6,6 @@ coverage.html
logs
testdata
/app/demo/pullds2rtmp
/pre-commit.sh
/coverage.txt
/TODO.md

@ -70,9 +70,9 @@ Play multi protocol stream from lalserver via ffplay:
```shell
$ffplay rtmp://127.0.0.1/live/test110
$ffplay http://127.0.0.1:8080/live/test110.flv
$ffplay http://127.0.0.1:8081/hls/test110/playlist.m3u8
$ffplay http://127.0.0.1:8081/hls/test110/record.m3u8
$ffplay http://127.0.0.1:8082/live/test110.ts
$ffplay http://127.0.0.1:8080/hls/test110/playlist.m3u8
$ffplay http://127.0.0.1:8080/hls/test110/record.m3u8
$ffplay http://127.0.0.1:8080/live/test110.ts
```
## More than a server, act as package and client

@ -1,68 +1,82 @@
```
{
"# doc of config": "https://pengrl.com/lal/#/ConfigBrief", // 配置文件对应的文档说明链接,在程序中没实际用途
"conf_version": "0.2.0", // 配置文件版本号,业务方不应该手动修改,程序中会检查该版本号是否与代码中声明的一致
"# doc of config": "https://pengrl.com/lal/#/ConfigBrief", //. 配置文件对应的文档说明链接,在程序中没实际用途
"conf_version": "0.2.2", //. 配置文件版本号,业务方不应该手动修改,程序中会检查该版本
// 号是否与代码中声明的一致
"rtmp": {
"enable": true, // 是否开启rtmp服务的监听
"addr": ":19350", // RTMP服务监听的端口客户端向lalserver推拉流都是这个地址
"gop_num": 2 // RTMP拉流的GOP缓存数量加速秒开
"enable": true, //. 是否开启rtmp服务的监听
"addr": ":19350", //. RTMP服务监听的端口客户端向lalserver推拉流都是这个地址
"gop_num": 2 //. RTMP拉流的GOP缓存数量加速秒开
},
"default_http": { // http监听相关的默认配置如果hls, httpflv, httpts中没有单独配置以下配置项则使用default_http中的配置
// 注意hls, httpflv, httpts服务是否开启不由此处决定
"http_listen_addr": ":8080", // HTTP监听地址
"https_listen_addr": ":4433", // HTTPS监听地址
"https_cert_file": "./conf/cert.pem", // HTTPS的本地cert文件地址
"https_key_file": "./conf/key.pem" // HTTPS的本地key文件地址
"default_http": { //. http监听相关的默认配置如果hls, httpflv, httpts中没有单独配置以下配置项
// 则使用default_http中的配置
// 注意hls, httpflv, httpts服务是否开启不由此处决定
"http_listen_addr": ":8080", //. HTTP监听地址
"https_listen_addr": ":4433", //. HTTPS监听地址
"https_cert_file": "./conf/cert.pem", //. HTTPS的本地cert文件地址
"https_key_file": "./conf/key.pem" //. HTTPS的本地key文件地址
},
"httpflv": {
"enable": true, // 是否开启HTTP-FLV服务的监听
"enable_https": true, // 是否开启HTTPS-FLV监听
"gop_num": 2
"enable": true, //. 是否开启HTTP-FLV服务的监听
"enable_https": true, //. 是否开启HTTPS-FLV监听
"url_pattern": "/live/", //. 拉流url路由地址。默认值`/live/`,对应`/live/{streamName}.flv`
"gop_num": 2 //.
},
"hls": {
"enable": true, // 是否开启HLS服务的监听
"out_path": "/tmp/lal/hls/", // HLS文件保存根目录
"fragment_duration_ms": 3000, // 单个TS文件切片时长单位毫秒
"fragment_num": 6, // m3u8文件列表中ts文件的数量
"cleanup_mode": 1, // HLS文件清理模式
// 0 不删除m3u8+ts文件可用于录制等场景
// 1 在输入流结束后删除m3u8+ts文件
// 注意,确切的删除时间是推流结束后的<fragment_duration_ms> * <fragment_num> * 2的时间点
// 推迟一小段时间删除是为了避免输入流刚结束HLS的拉流端还没有拉取完
// 2 推流过程中持续删除过期的ts文件只保留最近的<fragment_num> * 2个左右的ts文件
"use_memory_as_disk_flag": false // 是否使用内存取代磁盘保存m3u8+ts文件
"enable": true, //. 是否开启HLS服务的监听
"enable_https": true, //. 是否开启HTTPS-FLV监听
"url_pattern": "/hls/", //. 拉流url路由地址默认值`/hls/`,对应:
// - `/hls/{streamName}.m3u8`
// `/hls/{streamName}/playlist.m3u8`
// `/hls/{streamName}/record.m3u8`
// - `/hls/{streamName}/{streamName}-{timestamp}-{index}.ts`
// `/hls/{streamName}-{timestamp}-{index}.ts`
// 注意hls的url_pattern不能和httpflv、httpts的url_pattern相同
"out_path": "/tmp/lal/hls/", //. HLS文件保存根目录
"fragment_duration_ms": 3000, //. 单个TS文件切片时长单位毫秒
"fragment_num": 6, //. m3u8文件列表中ts文件的数量
"cleanup_mode": 1, //. HLS文件清理模式
// 0 不删除m3u8+ts文件可用于录制等场景
// 1 在输入流结束后删除m3u8+ts文件
// 注意,确切的删除时间是推流结束后的<fragment_duration_ms> * <fragment_num> * 2
// 的时间点
// 推迟一小段时间删除是为了避免输入流刚结束HLS的拉流端还没有拉取完
// 2 推流过程中持续删除过期的ts文件只保留最近的<fragment_num> * 2个左右的ts文件
"use_memory_as_disk_flag": false //. 是否使用内存取代磁盘保存m3u8+ts文件
},
"httpts": {
"enable": true // 是否开启HTTP-TS服务的监听。注意这并不是HLS中的TS而是在一条HTTP长连接上持续性传输TS流
"enable": true, //. 是否开启HTTP-TS服务的监听。注意这并不是HLS中的TS而是在一条HTTP长连接上持续性传输TS流
"enable_https": true, //. 是否开启HTTPS-FLV监听
"url_pattern": "/live/" //. 拉流url路由地址。默认值`/live/`,对应`/live/{streamName}.flv`
},
"rtsp": {
"enable": true, // 是否开启rtsp服务的监听目前只支持rtsp推流
"addr": ":5544" // rtsp推流地址
"enable": true, //. 是否开启rtsp服务的监听目前只支持rtsp推流
"addr": ":5544" //. rtsp推流地址
},
"record": {
"enable_flv": true, // 是否开启flv录制
"flv_out_path": "/tmp/lal/flv/", // flv录制目录
"enable_mpegts": true, // 是否开启mpegts录制。注意此处是长ts文件录制hls录制由上面的hls配置控制
"mpegts_out_path": "/tmp/lal/mpegts" // mpegts录制目录
"enable_flv": true, //. 是否开启flv录制
"flv_out_path": "/tmp/lal/flv/", //. flv录制目录
"enable_mpegts": true, //. 是否开启mpegts录制。注意此处是长ts文件录制hls录制由上面的hls配置控制
"mpegts_out_path": "/tmp/lal/mpegts" //. mpegts录制目录
},
"relay_push": {
"enable": false, // 是否开启中继转推功能,开启后,自身接收到的所有流都会转推出去
"addr_list":[ // 中继转推的对端地址支持填写多个地址做1对n的转推。格式举例 "127.0.0.1:19351"
"enable": false, //. 是否开启中继转推功能,开启后,自身接收到的所有流都会转推出去
"addr_list":[ //. 中继转推的对端地址支持填写多个地址做1对n的转推。格式举例 "127.0.0.1:19351"
]
},
"relay_pull": {
"enable": false, // 是否开启回源拉流功能,开启后,当自身接收到拉流请求,而流不存在时,会从其他服务器拉取这个流到本地
"addr": "" // 回源拉流的地址。格式举例 "127.0.0.1:19351"
"enable": false, //. 是否开启回源拉流功能,开启后,当自身接收到拉流请求,而流不存在时,会从其他服务器拉取这个流到本地
"addr": "" //. 回源拉流的地址。格式举例 "127.0.0.1:19351"
},
"http_api": {
"enable": true, // 是否开启HTTP API接口
"addr": ":8083" // 监听地址
"enable": true, //. 是否开启HTTP API接口
"addr": ":8083" //. 监听地址
},
"server_id": "1", // 当前lalserver唯一ID。多个lalserver HTTP Notify同一个地址时可通过该ID区分
"server_id": "1", //. 当前lalserver唯一ID。多个lalserver HTTP Notify同一个地址时可通过该ID区分
"http_notify": {
"enable": true, // 是否开启HTTP Notify事件回调
"update_interval_sec": 5, // update事件回调间隔单位毫秒
"on_server_start": "http://127.0.0.1:10101/on_server_start", // 各事件HTTP Notify事件回调地址
"enable": true, //. 是否开启HTTP Notify事件回调
"update_interval_sec": 5, //. update事件回调间隔单位毫秒
"on_server_start": "http://127.0.0.1:10101/on_server_start", //. 各事件HTTP Notify事件回调地址
"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",
@ -71,16 +85,16 @@
"on_rtmp_connect": "http://127.0.0.1:10101/on_rtmp_connect"
},
"pprof": {
"enable": true, // 是否开启Go pprof web服务的监听
"addr": ":8084" // Go pprof web地址
"enable": true, //. 是否开启Go pprof web服务的监听
"addr": ":8084" //. Go pprof web地址
},
"log": {
"level": 1, // 日志级别0 trace, 1 debug, 2 info, 3 warn, 4 error, 5 fatal
"filename": "./logs/lalserver.log", // 日志输出文件
"is_to_stdout": true, // 是否打印至标志控制台输出
"is_rotate_daily": true, // 日志按天翻滚
"short_file_flag": true, // 日志末尾是否携带源码文件名以及行号的信息
"assert_behavior": 1 // 日志断言的行为1 只打印错误日志 2 打印并退出程序 3 打印并panic
"level": 1, //. 日志级别0 trace, 1 debug, 2 info, 3 warn, 4 error, 5 fatal
"filename": "./logs/lalserver.log", //. 日志输出文件
"is_to_stdout": true, //. 是否打印至标志控制台输出
"is_rotate_daily": true, //. 日志按天翻滚
"short_file_flag": true, //. 日志末尾是否携带源码文件名以及行号的信息
"assert_behavior": 1 //. 日志断言的行为1 只打印错误日志 2 打印并退出程序 3 打印并panic
}
}
```

@ -1,10 +1,10 @@
{
"# doc of config": "https://pengrl.com/lal/#/ConfigBrief",
"conf_version": "v0.2.1",
"conf_version": "v0.2.2",
"rtmp": {
"enable": true,
"addr": ":1935",
"gop_num": 2,
"gop_num": 0,
"merge_write_size": 0
},
"default_http": {
@ -16,11 +16,13 @@
"httpflv": {
"enable": true,
"enable_https": false,
"gop_num": 2
"url_pattern": "/live/",
"gop_num": 0
},
"hls": {
"enable": true,
"enable_https": false,
"url_pattern": "/hls/",
"out_path": "/tmp/lal/hls/",
"fragment_duration_ms": 3000,
"fragment_num": 6,
@ -29,7 +31,8 @@
},
"httpts": {
"enable": true,
"enable_https":false
"enable_https":false,
"url_pattern": "/live/"
},
"rtsp": {
"enable": true,

@ -1,10 +1,11 @@
{
"# doc of config": "https://pengrl.com/lal/#/ConfigBrief",
"conf_version": "v0.2.0",
"conf_version": "v0.2.2",
"rtmp": {
"enable": true,
"addr": ":1935",
"gop_num": 2
"gop_num": 0,
"merge_write_size": 0
},
"default_http": {
"http_listen_addr": ":8080",
@ -15,11 +16,13 @@
"httpflv": {
"enable": true,
"enable_https": false,
"gop_num": 2
"url_pattern": "/live/",
"gop_num": 0
},
"hls": {
"enable": true,
"enable_https": false,
"url_pattern": "/hls/",
"out_path": "/tmp/lal/hls/",
"fragment_duration_ms": 3000,
"fragment_num": 6,
@ -28,7 +31,8 @@
},
"httpts": {
"enable": true,
"enable_https":false
"enable_https":false,
"url_pattern": "/live/"
},
"rtsp": {
"enable": true,

@ -33,6 +33,10 @@ func SetUseMemoryAsDiskFlag(flag bool) {
})
}
func ReadFile(filename string) ([]byte, error) {
return fslCtx.ReadFile(filename)
}
func RemoveAll(path string) error {
return fslCtx.RemoveAll(path)
}

@ -95,9 +95,9 @@ type fragmentInfo struct {
// @param observer 可以为nil如果不为nilTS流将回调给上层
func NewMuxer(streamName string, enable bool, config *MuxerConfig, observer MuxerObserver) *Muxer {
uk := base.GenUKHLSMuxer()
op := getMuxerOutPath(config.OutPath, streamName)
playlistFilename := getM3U8Filename(op, streamName)
recordPlaylistFilename := getRecordM3U8Filename(op, streamName)
op := PathStrategy.GetMuxerOutPath(config.OutPath, streamName)
playlistFilename := PathStrategy.GetLiveM3U8FileName(op, streamName)
recordPlaylistFilename := PathStrategy.GetRecordM3U8FileName(op, streamName)
playlistFilenameBak := fmt.Sprintf("%s.bak", playlistFilename)
recordPlaylistFilenameBak := fmt.Sprintf("%s.bak", recordPlaylistFilename)
frags := make([]fragmentInfo, 2*config.FragmentNum+1)
@ -281,8 +281,8 @@ func (m *Muxer) openFragment(ts uint64, discont bool) error {
id := m.getFragmentID()
filename := getTSFilename(m.streamName, id, int(time.Now().Unix()))
filenameWithPath := getTSFilenameWithPath(m.outPath, filename)
filename := PathStrategy.GetTSFileName(m.streamName, id, int(time.Now().UnixNano()/1e6))
filenameWithPath := PathStrategy.GetTSFileNameWithPath(m.outPath, filename)
if m.enable {
if err := m.fragment.OpenFile(filenameWithPath); err != nil {
return err
@ -337,7 +337,7 @@ func (m *Muxer) closeFragment(isLast bool) error {
//
frag := m.getCurrFrag()
if frag.filename != "" {
filenameWithPath := getTSFilenameWithPath(m.outPath, frag.filename)
filenameWithPath := PathStrategy.GetTSFileNameWithPath(m.outPath, frag.filename)
if err := fslCtx.Remove(filenameWithPath); err != nil {
nazalog.Warnf("[%s] remove stale fragment file failed. filename=%s, err=%+v", m.UniqueKey, filenameWithPath, err)
}

@ -1,88 +0,0 @@
// 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 hls
import (
"fmt"
"path/filepath"
"strings"
)
// 本文件聚合以下功能:
// - 生成HLSm3u8文件+ts文件文件命名规则以及文件存放规则
// - HTTP请求HLS时request URI和文件路径的映射规则
// HTTP请求URI格式已经文件路径的映射规则
//
// 假设
// 流名称="test110"
// rootPath="/tmp/lal/hls/"
//
// 则
// http://127.0.0.1:8081/hls/test110/playlist.m3u8 -> /tmp/lal/hls/test110/playlist.m3u8
// http://127.0.0.1:8081/hls/test110/record.m3u8 -> /tmp/lal/hls/test110/record.m3u8
// http://127.0.0.1:8081/hls/test110/timestamp-0.ts -> /tmp/lal/hls/test110/timestamp-0.ts
type requestInfo struct {
fileName string
streamName string
fileType string
}
// RequestURI example:
// uri -> fileName streamName fileType
// http://127.0.0.1:8081/hls/test110/playlist.m3u8 -> playlist.m3u8 test110 m3u8
// http://127.0.0.1:8081/hls/test110/record.m3u8 -> record.m3u8 test110 m3u8
// http://127.0.0.1:8081/hls/test110/timestamp-0.ts -> timestamp-0.ts test110 ts
func parseRequestInfo(uri string) (ri requestInfo) {
ss := strings.Split(uri, "/")
if len(ss) < 2 {
return
}
ri.streamName = ss[len(ss)-2]
ri.fileName = ss[len(ss)-1]
ss = strings.Split(ri.fileName, ".")
if len(ss) < 2 {
return
}
ri.fileType = ss[len(ss)-1]
return
}
// <rootOutPath>/<ri.streamName>/<ri.fileName>
func readFileContent(rootOutPath string, ri requestInfo) ([]byte, error) {
filename := filepath.Join(rootOutPath, ri.streamName, ri.fileName)
return fslCtx.ReadFile(filename)
}
// <rootOutPath>/<streamName>
func getMuxerOutPath(rootOutPath string, streamName string) string {
return filepath.Join(rootOutPath, streamName)
}
// @param outPath 参考func getMuxerOutPath
func getM3U8Filename(outPath string, streamName string) string {
return filepath.Join(outPath, "playlist.m3u8")
}
// @param outPath 参考func getMuxerOutPath
func getRecordM3U8Filename(outPath string, streamName string) string {
return filepath.Join(outPath, "record.m3u8")
}
// @param outPath 参考func getMuxerOutPath
func getTSFilenameWithPath(outpath string, filename string) string {
return filepath.Join(outpath, filename)
}
func getTSFilename(streamName string, id int, timestamp int) string {
return fmt.Sprintf("%d-%d.ts", timestamp, id)
}

@ -0,0 +1,155 @@
// 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 hls
import (
"fmt"
"path/filepath"
"strings"
)
// 聚合以下功能:
// - 落盘策略: 生成HLSm3u8文件+ts文件文件命名规则以及文件存放规则
// - 路由策略: HTTP请求HLS时request URI和文件路径的映射规则
type RequestInfo struct {
FileName string
FileType string
StreamName string
FileNameWithPath string
}
type IPathStrategy interface {
IPathRequestStrategy
IPathWriteStrategy
}
// 路由策略
// 接到HTTP请求时对应文件路径的映射逻辑
type IPathRequestStrategy interface {
// 解析HTTP请求得到文件名、文件类型、流名称、文件所在路径
GetRequestInfo(uri string, rootOutPath string) RequestInfo
}
// 落盘策略
type IPathWriteStrategy interface {
// 获取单个流对应的文件根路径
GetMuxerOutPath(rootOutPath string, streamName string) string
// 获取单个流对应的m3u8文件路径
//
// @param outPath: func GetMuxerOutPath的结果
GetLiveM3U8FileName(outPath string, streamName string) string
// 获取单个流对应的record类型的m3u8文件路径
//
// live m3u8和record m3u8的区别
// live记录的是当前最近的可播放内容record记录的是从流开始时的可播放内容
//
// @param outPath: func GetMuxerOutPath的结果
GetRecordM3U8FileName(outPath string, streamName string) string
// 获取单个流对应的ts文件路径
//
// @param outPath: func GetMuxerOutPath的结果
GetTSFileNameWithPath(outPath string, fileName string) string
// ts文件名的生成策略
GetTSFileName(streamName string, index int, timestamp int) string
}
// ---------------------------------------------------------------------------------------------------------------------
const (
playlistM3u8FileName = "playlist.m3u8"
recordM3u8FileName = "record.m3u8"
)
// 默认的路由,落盘策略
//
// 每个流在<rootPath>下以流名称生成一个子目录,目录下包含:
//
// - playlist.m3u8 实时的HLS文件定期刷新写入当前最新的TS文件列表淘汰过期的TS文件列表
// - record.m3u8 录制回放的HLS文件包含了从流开始至今的所有TS文件
// - test110-1620540712084-0.ts TS分片文件命名格式为{liveid}-{timestamp}-{index}.ts
// - test110-1620540716095-1.ts
// - ... 一系列的TS文件
//
//
// 假设
// 流名称="test110"
// rootPath="/tmp/lal/hls/"
//
// 则
// http://127.0.0.1:8080/hls/test110/playlist.m3u8 -> /tmp/lal/hls/test110/playlist.m3u8
// http://127.0.0.1:8080/hls/test110/record.m3u8 -> /tmp/lal/hls/test110/record.m3u8
// http://127.0.0.1:8080/hls/test110/test110-1620540712084-0.ts -> /tmp/lal/hls/test110/test110-1620540712084-0.ts
//
// http://127.0.0.1:8080/hls/test110.m3u8 -> /tmp/lal/hls/test110/playlist.m3u8
// http://127.0.0.1:8080/hls/test110-1620540712084-0.ts -> /tmp/lal/hls/test110/test110-1620540712084-0.ts
// 最下面这两个做了特殊映射
//
type DefaultPathStrategy struct {
}
// RequestURI example:
// uri -> FileName StreamName FileType FileNameWithPath
// /hls/test110.m3u8 -> test110.m3u8 test110 m3u8 {rootOutPath}/test110/playlist.m3u8
// /hls/test110/playlist.m3u8 -> playlist.m3u8 test110 m3u8 {rootOutPath}/test110/playlist.m3u8
// /hls/test110/record.m3u8 -> record.m3u8 test110 m3u8 {rootOutPath}/test110/record.m3u8
// /hls/test110/test110-1620540712084-.ts -> test110-1620540712084-.ts test110 ts {rootOutPath/test110/test110-1620540712084-.ts
// /hls/test110-1620540712084-.ts -> test110-1620540712084-.ts test110 ts {rootOutPath/test110/test110-1620540712084-.ts
func (dps *DefaultPathStrategy) GetRequestInfo(uri string, rootOutPath string) (ri RequestInfo) {
uriItems := strings.Split(uri, "/")
ri.FileName = uriItems[len(uriItems)-1]
fileNameItems := strings.Split(ri.FileName, ".")
fileNameWithOutType := fileNameItems[0]
ri.FileType = fileNameItems[len(fileNameItems)-1]
if ri.FileType == "m3u8" {
if ri.FileName == playlistM3u8FileName || ri.FileName == recordM3u8FileName {
ri.StreamName = uriItems[len(uriItems)-2]
ri.FileNameWithPath = filepath.Join(rootOutPath, ri.StreamName, ri.FileName)
} else {
ri.StreamName = fileNameWithOutType
ri.FileNameWithPath = filepath.Join(rootOutPath, ri.StreamName, playlistM3u8FileName)
}
} else if ri.FileType == "ts" {
ri.StreamName = dps.getStreamNameFromTSFileName(ri.FileName)
ri.FileNameWithPath = filepath.Join(rootOutPath, ri.StreamName, ri.FileName)
}
return
}
// <rootOutPath>/<streamName>
func (*DefaultPathStrategy) GetMuxerOutPath(rootOutPath string, streamName string) string {
return filepath.Join(rootOutPath, streamName)
}
func (*DefaultPathStrategy) GetLiveM3U8FileName(outPath string, streamName string) string {
return filepath.Join(outPath, playlistM3u8FileName)
}
func (*DefaultPathStrategy) GetRecordM3U8FileName(outPath string, streamName string) string {
return filepath.Join(outPath, recordM3u8FileName)
}
func (*DefaultPathStrategy) GetTSFileNameWithPath(outPath string, fileName string) string {
return filepath.Join(outPath, fileName)
}
func (*DefaultPathStrategy) GetTSFileName(streamName string, index int, timestamp int) string {
return fmt.Sprintf("%s-%d-%d.ts", streamName, timestamp, index)
}
func (*DefaultPathStrategy) getStreamNameFromTSFileName(fileName string) string {
return strings.Split(fileName, "-")[0]
}

@ -0,0 +1,59 @@
// 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 hls_test
import (
"testing"
"github.com/q191201771/lal/pkg/hls"
"github.com/q191201771/naza/pkg/assert"
)
func TestDefaultPathStrategy_GetRequestInfo(t *testing.T) {
dps := &hls.DefaultPathStrategy{}
rootOutPath := "/tmp/lal/hls/"
golden := map[string]hls.RequestInfo{
"/hls/test110.m3u8": {
FileName: "test110.m3u8",
FileType: "m3u8",
StreamName: "test110",
FileNameWithPath: "/tmp/lal/hls/test110/playlist.m3u8",
},
"/hls/test110/playlist.m3u8": {
FileName: "playlist.m3u8",
FileType: "m3u8",
StreamName: "test110",
FileNameWithPath: "/tmp/lal/hls/test110/playlist.m3u8",
},
"/hls/test110/record.m3u8": {
FileName: "record.m3u8",
FileType: "m3u8",
StreamName: "test110",
FileNameWithPath: "/tmp/lal/hls/test110/record.m3u8",
},
"/hls/test110/test110-1620540712084-0.ts": {
FileName: "test110-1620540712084-0.ts",
FileType: "ts",
StreamName: "test110",
FileNameWithPath: "/tmp/lal/hls/test110/test110-1620540712084-0.ts",
},
"/hls/test110-1620540712084-0.ts": {
FileName: "test110-1620540712084-0.ts",
FileType: "ts",
StreamName: "test110",
FileNameWithPath: "/tmp/lal/hls/test110/test110-1620540712084-0.ts",
},
}
for k, v := range golden {
out := dps.GetRequestInfo(k, rootOutPath)
assert.Equal(t, v, out)
}
}

@ -55,23 +55,23 @@ func (s *ServerHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
// TODO chef:
// - check appname in URI path
ri := parseRequestInfo(req.RequestURI)
ri := PathStrategy.GetRequestInfo(req.RequestURI, s.outPath)
//nazalog.Debugf("%+v", ri)
if ri.fileName == "" || ri.streamName == "" || (ri.fileType != "m3u8" && ri.fileType != "ts") {
nazalog.Warnf("invalid hls request. request=%+v", ri)
if ri.FileName == "" || ri.StreamName == "" || ri.FileNameWithPath == "" || (ri.FileType != "m3u8" && ri.FileType != "ts") {
nazalog.Warnf("invalid hls request. uri=%s, request=%+v", req.RequestURI, ri)
resp.WriteHeader(404)
return
}
content, err := readFileContent(s.outPath, ri)
content, err := ReadFile(ri.FileNameWithPath)
if err != nil {
nazalog.Warnf("read hls file failed. request=%+v, err=%+v", ri, err)
resp.WriteHeader(404)
return
}
switch ri.fileType {
switch ri.FileType {
case "m3u8":
resp.Header().Add("Content-Type", "application/x-mpegurl")
resp.Header().Add("Server", base.LALHLSM3U8Server)

@ -8,6 +8,10 @@
package hls
var (
PathStrategy IPathStrategy = &DefaultPathStrategy{}
)
var (
calcFragmentHeaderQueueSize = 16
)

@ -67,6 +67,7 @@ func NewPullSession(modOptions ...ModPullSessionOption) *PullSession {
return s
}
// @param tag: 底层保证回调上来的Raw数据长度是完整的但是不会分析Raw内部的编码数据
type OnReadFLVTag func(tag Tag)
// 阻塞直到和对端完成拉流前握手部分的工作也即发送完HTTP Request或者发生错误

@ -9,11 +9,26 @@
package logic
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
"strings"
"github.com/q191201771/lal/pkg/base"
"github.com/q191201771/lal/pkg/hls"
"github.com/q191201771/naza/pkg/nazajson"
"github.com/q191201771/naza/pkg/nazalog"
)
const ConfVersion = "v0.2.1"
const ConfVersion = "v0.2.2"
const (
defaultHLSCleanupMode = hls.CleanupModeInTheEnd
defaultHTTPFLVURLPattern = "/live/"
defaultHTTPTSURLPattern = "/live/"
defaultHLSURLPattern = "/hls/"
)
type Config struct {
ConfVersion string `json:"conf_version"`
@ -109,8 +124,9 @@ type PProfConfig struct {
type CommonHTTPServerConfig struct {
CommonHTTPAddrConfig
Enable bool `json:"enable"`
EnableHTTPS bool `json:"enable_https"`
Enable bool `json:"enable"`
EnableHTTPS bool `json:"enable_https"`
URLPattern string `json:"url_pattern"`
}
type CommonHTTPAddrConfig struct {
@ -119,3 +135,186 @@ type CommonHTTPAddrConfig struct {
HTTPSCertFile string `json:"https_cert_file"`
HTTPSKeyFile string `json:"https_key_file"`
}
func LoadConfAndInitLog(confFile string) *Config {
// 读取配置文件并解析原始内容
rawContent, err := ioutil.ReadFile(confFile)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "read conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
if err = json.Unmarshal(rawContent, &config); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "unmarshal conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
j, err := nazajson.New(rawContent)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "nazajson unmarshal conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
// 初始化日志,注意,这一步尽量提前,使得后续的日志内容按我们的日志配置输出
// 日志配置项不存在时,设置默认值
if !j.Exist("log.level") {
config.LogConfig.Level = nazalog.LevelDebug
}
if !j.Exist("log.filename") {
config.LogConfig.Filename = "./logs/lalserver.log"
}
if !j.Exist("log.is_to_stdout") {
config.LogConfig.IsToStdout = true
}
if !j.Exist("log.is_rotate_daily") {
config.LogConfig.IsRotateDaily = true
}
if !j.Exist("log.short_file_flag") {
config.LogConfig.ShortFileFlag = true
}
if !j.Exist("log.timestamp_flag") {
config.LogConfig.TimestampFlag = true
}
if !j.Exist("log.timestamp_with_ms_flag") {
config.LogConfig.TimestampWithMSFlag = true
}
if !j.Exist("log.level_flag") {
config.LogConfig.LevelFlag = true
}
if !j.Exist("log.assert_behavior") {
config.LogConfig.AssertBehavior = nazalog.AssertError
}
if err := nazalog.Init(func(option *nazalog.Option) {
*option = config.LogConfig
}); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "initial log failed. err=%+v\n", err)
base.OSExitAndWaitPressIfWindows(1)
}
nazalog.Info("initial log succ.")
// 打印Logo
nazalog.Info(`
__ ___ __
/ / / | / /
/ / / /| | / /
/ /___/ ___ |/ /___
/_____/_/ |_/_____/
`)
// 检查配置版本号是否匹配
if config.ConfVersion != ConfVersion {
nazalog.Warnf("config version invalid. conf version of lalserver=%s, conf version of config file=%s",
ConfVersion, config.ConfVersion)
}
// 检查一级配置项
keyFieldList := []string{
"rtmp",
"httpflv",
"hls",
"httpts",
"rtsp",
"record",
"relay_push",
"relay_pull",
"http_api",
"http_notify",
"pprof",
"log",
}
for _, kf := range keyFieldList {
if !j.Exist(kf) {
nazalog.Warnf("missing config item %s", kf)
}
}
// 如果具体的HTTP应用没有设置HTTP监听相关的配置则尝试使用全局配置
mergeCommonHTTPAddrConfig(&config.HTTPFLVConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
mergeCommonHTTPAddrConfig(&config.HTTPTSConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
mergeCommonHTTPAddrConfig(&config.HLSConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
// 配置不存在时,设置默认值
if (config.HLSConfig.Enable || config.HLSConfig.EnableHTTPS) && !j.Exist("hls.cleanup_mode") {
nazalog.Warnf("config hls.cleanup_mode not exist. set to default which is %d", defaultHLSCleanupMode)
config.HLSConfig.CleanupMode = defaultHLSCleanupMode
}
if (config.HTTPFLVConfig.Enable || config.HTTPFLVConfig.EnableHTTPS) && !j.Exist("httpflv.url_pattern") {
nazalog.Warnf("config httpflv.url_pattern not exist. set to default wchich is %s", defaultHTTPFLVURLPattern)
config.HTTPFLVConfig.URLPattern = defaultHTTPFLVURLPattern
}
if (config.HTTPTSConfig.Enable || config.HTTPTSConfig.EnableHTTPS) && !j.Exist("httpts.url_pattern") {
nazalog.Warnf("config httpts.url_pattern not exist. set to default wchich is %s", defaultHTTPTSURLPattern)
config.HTTPTSConfig.URLPattern = defaultHTTPTSURLPattern
}
if (config.HLSConfig.Enable || config.HLSConfig.EnableHTTPS) && !j.Exist("hls.url_pattern") {
nazalog.Warnf("config hls.url_pattern not exist. set to default wchich is %s", defaultHLSURLPattern)
config.HTTPFLVConfig.URLPattern = defaultHLSURLPattern
}
// 对一些常见的格式错误做修复
// 确保url pattern以`/`开始,并以`/`结束
if urlPattern, changed := ensureStartAndEndWithSlash(config.HTTPFLVConfig.URLPattern); changed {
nazalog.Warnf("fix config. httpflv.url_pattern %s -> %s", config.HTTPFLVConfig.URLPattern, urlPattern)
config.HTTPFLVConfig.URLPattern = urlPattern
}
if urlPattern, changed := ensureStartAndEndWithSlash(config.HTTPTSConfig.URLPattern); changed {
nazalog.Warnf("fix config. httpts.url_pattern %s -> %s", config.HTTPTSConfig.URLPattern, urlPattern)
config.HTTPFLVConfig.URLPattern = urlPattern
}
if urlPattern, changed := ensureStartAndEndWithSlash(config.HLSConfig.URLPattern); changed {
nazalog.Warnf("fix config. hls.url_pattern %s -> %s", config.HLSConfig.URLPattern, urlPattern)
config.HTTPFLVConfig.URLPattern = urlPattern
}
// 把配置文件原始内容中的换行去掉,使得打印日志时紧凑一些
lines := strings.Split(string(rawContent), "\n")
if len(lines) == 1 {
lines = strings.Split(string(rawContent), "\r\n")
}
var tlines []string
for _, l := range lines {
tlines = append(tlines, strings.TrimSpace(l))
}
compactRawContent := strings.Join(tlines, " ")
nazalog.Infof("load conf file succ. filename=%s, raw content=%s parsed=%+v", confFile, compactRawContent, config)
return config
}
func mergeCommonHTTPAddrConfig(dst, src *CommonHTTPAddrConfig) {
if dst.HTTPListenAddr == "" && src.HTTPListenAddr != "" {
dst.HTTPListenAddr = src.HTTPListenAddr
}
if dst.HTTPSListenAddr == "" && src.HTTPSListenAddr != "" {
dst.HTTPSListenAddr = src.HTTPSListenAddr
}
if dst.HTTPSCertFile == "" && src.HTTPSCertFile != "" {
dst.HTTPSCertFile = src.HTTPSCertFile
}
if dst.HTTPSKeyFile == "" && src.HTTPSKeyFile != "" {
dst.HTTPSKeyFile = src.HTTPSKeyFile
}
}
func ensureStartWithSlash(in string) (out string, changed bool) {
if in == "" {
return in, false
}
if in[0] == '/' {
return in, false
}
return "/" + in, true
}
func ensureEndWithSlash(in string) (out string, changed bool) {
if in == "" {
return in, false
}
if in[len(in)-1] == '/' {
return in, false
}
return in + "/", true
}
func ensureStartAndEndWithSlash(in string) (out string, changed bool) {
n, c := ensureStartWithSlash(in)
n2, c2 := ensureEndWithSlash(n)
return n2, c || c2
}

@ -9,18 +9,13 @@
package logic
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
_ "net/http/pprof"
"os"
"strings"
"github.com/q191201771/lal/pkg/hls"
"github.com/q191201771/naza/pkg/nazajson"
"github.com/q191201771/lal/pkg/base"
"github.com/q191201771/lal/pkg/hls"
"github.com/q191201771/naza/pkg/bininfo"
"github.com/q191201771/naza/pkg/nazalog"
@ -81,123 +76,6 @@ func Dispose() {
sm.Dispose()
}
func LoadConfAndInitLog(confFile string) *Config {
// 读取配置文件并解析原始内容
rawContent, err := ioutil.ReadFile(confFile)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "read conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
if err = json.Unmarshal(rawContent, &config); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "unmarshal conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
j, err := nazajson.New(rawContent)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "nazajson unmarshal conf file failed. file=%s err=%+v", confFile, err)
base.OSExitAndWaitPressIfWindows(1)
}
// 初始化日志,注意,这一步尽量提前,使得后续的日志内容按我们的日志配置输出
// 日志配置项不存在时,设置默认值
if !j.Exist("log.level") {
config.LogConfig.Level = nazalog.LevelDebug
}
if !j.Exist("log.filename") {
config.LogConfig.Filename = "./logs/lalserver.log"
}
if !j.Exist("log.is_to_stdout") {
config.LogConfig.IsToStdout = true
}
if !j.Exist("log.is_rotate_daily") {
config.LogConfig.IsRotateDaily = true
}
if !j.Exist("log.short_file_flag") {
config.LogConfig.ShortFileFlag = true
}
if !j.Exist("log.timestamp_flag") {
config.LogConfig.TimestampFlag = true
}
if !j.Exist("log.timestamp_with_ms_flag") {
config.LogConfig.TimestampWithMSFlag = true
}
if !j.Exist("log.level_flag") {
config.LogConfig.LevelFlag = true
}
if !j.Exist("log.assert_behavior") {
config.LogConfig.AssertBehavior = nazalog.AssertError
}
if err := nazalog.Init(func(option *nazalog.Option) {
*option = config.LogConfig
}); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "initial log failed. err=%+v\n", err)
base.OSExitAndWaitPressIfWindows(1)
}
nazalog.Info("initial log succ.")
// 打印Logo
nazalog.Info(`
__ ___ __
/ / / | / /
/ / / /| | / /
/ /___/ ___ |/ /___
/_____/_/ |_/_____/
`)
// 检查配置版本号是否匹配
if config.ConfVersion != ConfVersion {
nazalog.Warnf("config version invalid. conf version of lalserver=%s, conf version of config file=%s",
ConfVersion, config.ConfVersion)
}
// 检查一级配置项
keyFieldList := []string{
"rtmp",
"httpflv",
"hls",
"httpts",
"rtsp",
"record",
"relay_push",
"relay_pull",
"http_api",
"http_notify",
"pprof",
"log",
}
for _, kf := range keyFieldList {
if !j.Exist(kf) {
nazalog.Warnf("missing config item %s", kf)
}
}
// 如果具体的HTTP应用没有设置HTTP监听相关的配置则尝试使用全局配置
mergeCommonHTTPAddrConfig(&config.HTTPFLVConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
mergeCommonHTTPAddrConfig(&config.HTTPTSConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
mergeCommonHTTPAddrConfig(&config.HLSConfig.CommonHTTPAddrConfig, &config.DefaultHTTPConfig.CommonHTTPAddrConfig)
// 配置不存在时,设置默认值
if !j.Exist("hls.cleanup_mode") {
const defaultMode = hls.CleanupModeInTheEnd
nazalog.Warnf("config hls.cleanup_mode not exist. default is %d", defaultMode)
config.HLSConfig.CleanupMode = defaultMode
}
// 把配置文件原始内容中的换行去掉,使得打印日志时紧凑一些
lines := strings.Split(string(rawContent), "\n")
if len(lines) == 1 {
lines = strings.Split(string(rawContent), "\r\n")
}
var tlines []string
for _, l := range lines {
tlines = append(tlines, strings.TrimSpace(l))
}
compactRawContent := strings.Join(tlines, " ")
nazalog.Infof("load conf file succ. filename=%s, raw content=%s parsed=%+v", confFile, compactRawContent, config)
return config
}
func runWebPProf(addr string) {
nazalog.Infof("start web pprof listen. addr=%s", addr)
@ -209,18 +87,3 @@ func runWebPProf(addr string) {
return
}
}
func mergeCommonHTTPAddrConfig(dst, src *CommonHTTPAddrConfig) {
if dst.HTTPListenAddr == "" && src.HTTPListenAddr != "" {
dst.HTTPListenAddr = src.HTTPListenAddr
}
if dst.HTTPSListenAddr == "" && src.HTTPSListenAddr != "" {
dst.HTTPSListenAddr = src.HTTPSListenAddr
}
if dst.HTTPSCertFile == "" && src.HTTPSCertFile != "" {
dst.HTTPSCertFile = src.HTTPSCertFile
}
if dst.HTTPSKeyFile == "" && src.HTTPSKeyFile != "" {
dst.HTTPSKeyFile = src.HTTPSKeyFile
}
}

@ -69,41 +69,41 @@ func NewServerManager() *ServerManager {
func (sm *ServerManager) RunLoop() error {
httpNotify.OnServerStart()
var addMux = func(config CommonHTTPServerConfig, pattern string, handler base.Handler, name string) error {
var addMux = func(config CommonHTTPServerConfig, handler base.Handler, name string) error {
if config.Enable {
err := sm.httpServerManager.AddListen(
base.LocalAddrCtx{Addr: config.HTTPListenAddr},
pattern,
config.URLPattern,
handler,
)
if err != nil {
nazalog.Infof("add http listen for %s failed. addr=%s, pattern=%s, err=%+v", name, config.HTTPListenAddr, pattern, err)
nazalog.Infof("add http listen for %s failed. addr=%s, pattern=%s, err=%+v", name, config.HTTPListenAddr, config.URLPattern, err)
return err
}
nazalog.Infof("add http listen for %s. addr=%s, pattern=%s", name, config.HTTPListenAddr, pattern)
nazalog.Infof("add http listen for %s. addr=%s, pattern=%s", name, config.HTTPListenAddr, config.URLPattern)
}
if config.EnableHTTPS {
err := sm.httpServerManager.AddListen(
base.LocalAddrCtx{IsHTTPS: true, Addr: config.HTTPSListenAddr, CertFile: config.HTTPSCertFile, KeyFile: config.HTTPSKeyFile},
pattern,
config.URLPattern,
handler,
)
if err != nil {
nazalog.Infof("add https listen for %s failed. addr=%s, pattern=%s, err=%+v", name, config.HTTPListenAddr, pattern, err)
nazalog.Infof("add https listen for %s failed. addr=%s, pattern=%s, err=%+v", name, config.HTTPListenAddr, config.URLPattern, err)
return err
}
nazalog.Infof("add https listen for %s. addr=%s, pattern=%s", name, config.HTTPSListenAddr, pattern)
nazalog.Infof("add https listen for %s. addr=%s, pattern=%s", name, config.HTTPSListenAddr, config.URLPattern)
}
return nil
}
if err := addMux(config.HTTPFLVConfig.CommonHTTPServerConfig, HTTPFLVURLPath, sm.httpServerHandler.ServeSubSession, "httpflv"); err != nil {
if err := addMux(config.HTTPFLVConfig.CommonHTTPServerConfig, sm.httpServerHandler.ServeSubSession, "httpflv"); err != nil {
return err
}
if err := addMux(config.HTTPTSConfig.CommonHTTPServerConfig, HTTPTSURLPath, sm.httpServerHandler.ServeSubSession, "httpts"); err != nil {
if err := addMux(config.HTTPTSConfig.CommonHTTPServerConfig, sm.httpServerHandler.ServeSubSession, "httpts"); err != nil {
return err
}
if err := addMux(config.HTTPTSConfig.CommonHTTPServerConfig, HLSURLPath, sm.hlsServerHandler.ServeHTTP, "hls"); err != nil {
if err := addMux(config.HLSConfig.CommonHTTPServerConfig, sm.hlsServerHandler.ServeHTTP, "hls"); err != nil {
return err
}
@ -124,17 +124,6 @@ func (sm *ServerManager) RunLoop() error {
}()
}
//if sm.hlsServer != nil {
// if err := sm.hlsServer.Listen(); err != nil {
// return err
// }
// go func() {
// if err := sm.hlsServer.RunLoop(); err != nil {
// nazalog.Error(err)
// }
// }()
//}
if sm.rtspServer != nil {
if err := sm.rtspServer.Listen(); err != nil {
return err

@ -8,12 +8,6 @@
package logic
var (
HTTPFLVURLPath = "/live/"
HTTPTSURLPath = "/live/"
HLSURLPath = "/hls/"
)
//var relayPushCheckIntervalMS = 1000
var relayPushTimeoutMS = 5000
var relayPushWriteAVTimeoutMS = 5000

@ -51,7 +51,7 @@ type IRTPUnpackerProtocol interface {
// 注意不支持带B帧的视频流pts和dts永远相同
// pkt.PayloadType base.AVPacketPTXXX
// pkt.Payload 如果是AAC返回的是raw frame一个AVPacket只包含一帧
// 如果是AVC或HEVC一个AVPacket可能包含多个NAL(受STAP-A影响)所以NAL前包含4字节的长度信息
// 如果是AVC或HEVC是AVCC格式每个NAL前包含4字节NAL的长度
// AAC引用的是接收到的RTP包中的内存块
// AVC或者HEVC是新申请的内存块回调结束后内部不再使用该内存块
type OnAVPacket func(pkt base.AVPacket)

Loading…
Cancel
Save