Compare commits

..

11 Commits

Author SHA1 Message Date
ydajiang
2f86fe9d39 Update README.md 2025-06-02 14:05:34 +08:00
ydajiang
0770fbac4c feat: 使用cdn引入flv.js替代本地文件 2025-06-02 13:55:24 +08:00
ydajiang
1c59608f1e fix: http-flv拉流prev tag size不正确问题 2025-06-01 22:02:55 +08:00
ydajiang
e49b7a833e fix: hls拉流发生crash问题 2025-06-01 15:25:45 +08:00
ydajiang
914daeed5e fix: 1078 unpacking crash 2025-06-01 11:48:47 +08:00
ydajiang
3e371c1ac7 feat: 支持1078流转GB28181 2025-05-31 21:06:33 +08:00
ydajiang
976fd12b4b feat: flv支持网页预览 2025-05-26 10:23:37 +08:00
ydajiang
c67f234d45 fix: 1078sim卡号的前导零 2025-05-26 10:17:57 +08:00
ydajiang
d71014ae7f fix: 减少packet计数发生crash问题 2025-05-17 22:50:18 +08:00
ydajiang
7486fc1491 refactor: 优化流订阅处理逻辑 2025-05-14 19:50:12 +08:00
ydajiang
24fc44f9c7 refactor: 分离解析推流和转发流 2025-05-14 16:54:43 +08:00
39 changed files with 2019 additions and 1167 deletions

View File

@@ -1,6 +1,6 @@
## 简介
基于GoLang实现的流媒体服务器支持RTMP、GB28181、1078推流输出rtmp/http-flv/ws-flv/webrtc/hls/rtsp等拉流协议。支持如下编码器和流协议
基于GoLang实现的流媒体服务器支持RTMP、GB28181、jt1078推流、jt1078转GB28181输出rtmp/http-flv/ws-flv/webrtc/hls/rtsp等拉流协议。支持如下编码器和流协议
| Codec\Stream | RTMP | FLV | HLS | RTC | RTSP |
| ------------ | ---- | --- | --- | --- | ---- |

151
api.go
View File

@@ -100,13 +100,13 @@ func startApiServer(addr string) {
apiServer.router.HandleFunc("/api/v1/source/close", filterRequestBodyParams(apiServer.OnSourceClose, &IDS{})) // 关闭推流源
apiServer.router.HandleFunc("/api/v1/sink/list", filterRequestBodyParams(apiServer.OnSinkList, &IDS{})) // 查询某个推流源下,所有的拉流端列表
apiServer.router.HandleFunc("/api/v1/sink/close", filterRequestBodyParams(apiServer.OnSinkClose, &IDS{})) // 关闭拉流端
apiServer.router.HandleFunc("/api/v1/sink/add", filterRequestBodyParams(apiServer.OnSinkAdd, &GBOffer{})) // 级联/广播/JT转GB
apiServer.router.HandleFunc("/api/v1/streams/statistics", nil) // 统计所有推拉流
if stream.AppConfig.GB28181.Enable {
apiServer.router.HandleFunc("/ws/v1/gb28181/talk", apiServer.OnGBTalk) // 对讲的主讲人WebSocket连接
apiServer.router.HandleFunc("/api/v1/gb28181/offer/create", filterRequestBodyParams(apiServer.OnGBOfferCreate, &SourceSDP{}))
apiServer.router.HandleFunc("/api/v1/gb28181/answer/create", filterRequestBodyParams(apiServer.OnGBAnswerCreate, &GBOffer{}))
apiServer.router.HandleFunc("/api/v1/gb28181/source/create", filterRequestBodyParams(apiServer.OnGBOfferCreate, &SourceSDP{}))
apiServer.router.HandleFunc("/api/v1/gb28181/answer/set", filterRequestBodyParams(apiServer.OnGBSourceConnect, &SourceSDP{})) // active拉流模式下, 设置对方的地址
}
@@ -139,7 +139,7 @@ func (api *ApiServer) generateSinkID(remoteAddr string) stream.SinkID {
panic(err)
}
return stream.NetAddr2SinkId(tcpAddr)
return stream.NetAddr2SinkID(tcpAddr)
}
func (api *ApiServer) onFlv(sourceId string, w http.ResponseWriter, r *http.Request) {
@@ -163,27 +163,25 @@ func (api *ApiServer) onFlv(sourceId string, w http.ResponseWriter, r *http.Requ
func (api *ApiServer) onWSFlv(sourceId string, w http.ResponseWriter, r *http.Request) {
conn, err := api.upgrader.Upgrade(w, r, nil)
if err != nil {
log.Sugar.Errorf("websocket头检查失败 err:%s", err.Error())
log.Sugar.Errorf("ws拉流失败 source: %s err: %s", sourceId, err.Error())
w.WriteHeader(http.StatusBadRequest)
return
}
sink := flv.NewFLVSink(api.generateSinkID(r.RemoteAddr), sourceId, flv.NewWSConn(conn))
sink.SetUrlValues(r.URL.Query())
log.Sugar.Infof("ws-flv 连接 sink:%s", sink.String())
_, state := stream.PreparePlaySink(sink)
if utils.HookStateOK != state {
log.Sugar.Warnf("ws-flv 播放失败 sink:%s", sink.String())
w.WriteHeader(http.StatusForbidden)
return
ok := stream.SubscribeStream(sink, r.URL.Query())
if utils.HookStateOK != ok {
log.Sugar.Warnf("ws-flv 拉流失败 source: %s sink: %s", sourceId, sink.String())
_ = conn.Close()
} else {
log.Sugar.Infof("ws-flv 拉流成功 source: %s sink: %s", sourceId, sink.String())
}
netConn := conn.NetConn()
bytes := make([]byte, 64)
for {
if _, err := netConn.Read(bytes); err != nil {
log.Sugar.Infof("ws-flv 断开连接 sink:%s", sink.String())
log.Sugar.Infof("ws-flv 断开连接 source: %s sink:%s", sourceId, sink.String())
sink.Close()
break
}
@@ -195,29 +193,28 @@ func (api *ApiServer) onHttpFLV(sourceId string, w http.ResponseWriter, r *http.
w.Header().Set("Connection", "Keep-Alive")
w.Header().Set("Transfer-Encoding", "chunked")
hj, ok := w.(http.Hijacker)
if !ok {
var conn net.Conn
if hj, ok := w.(http.Hijacker); !ok {
log.Sugar.Errorf("http-flv 拉流失败 不支持hijacking. source: %s remote: %s", sourceId, r.RemoteAddr)
http.Error(w, "webserver doesn't support hijacking", http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusOK)
conn, _, err := hj.Hijack()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
} else {
w.WriteHeader(http.StatusOK)
var err error
if conn, _, err = hj.Hijack(); err != nil {
log.Sugar.Errorf("http-flv 拉流失败 source: %s remote: %s err: %s", sourceId, r.RemoteAddr, err.Error())
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
sink := flv.NewFLVSink(api.generateSinkID(r.RemoteAddr), sourceId, conn)
sink.SetUrlValues(r.URL.Query())
log.Sugar.Infof("http-flv 连接 sink:%s", sink.String())
_, state := stream.PreparePlaySink(sink)
if utils.HookStateOK != state {
log.Sugar.Warnf("http-flv 播放失败 sink:%s", sink.String())
w.WriteHeader(http.StatusForbidden)
return
ok := stream.SubscribeStream(sink, r.URL.Query())
if utils.HookStateOK != ok {
log.Sugar.Warnf("http-flv 拉流失败 source: %s sink: %s", sourceId, sink.String())
sink.Close()
} else {
log.Sugar.Infof("http-flv 拉流成功 source: %s sink: %s", sourceId, sink.String())
}
bytes := make([]byte, 64)
@@ -234,7 +231,7 @@ func (api *ApiServer) onTS(source string, w http.ResponseWriter, r *http.Request
sid := r.URL.Query().Get(hls.SessionIDKey)
var sink stream.Sink
if sid != "" {
sink = stream.SinkManager.Find(stream.SinkID(sid))
sink = hls.SinkManager.Find(stream.SinkID(sid))
}
if sink == nil {
log.Sugar.Errorf("hls session with id '%s' has expired.", sid)
@@ -255,7 +252,7 @@ func (api *ApiServer) onTS(source string, w http.ResponseWriter, r *http.Request
return
}
sink.(*hls.M3U8Sink).RefreshPlayTime()
sink.(*hls.M3U8Sink).RefreshPlayingTime()
w.Header().Set("Content-Type", "video/MP2T")
http.ServeFile(w, r, tsPath)
}
@@ -280,48 +277,34 @@ func (api *ApiServer) onHLS(source string, w http.ResponseWriter, r *http.Reques
return
}
sink := stream.SinkManager.Find(sid)
sink := hls.SinkManager.Find(sid)
if sink == nil {
// 创建sink
sink = hls.NewM3U8Sink(sid, source, sid)
sink.(*hls.M3U8Sink).RefreshPlayingTime()
if hls.SinkManager.Add(sink) {
ok := stream.SubscribeStream(sink, r.URL.Query())
if utils.HookStateOK != ok {
log.Sugar.Warnf("m3u8拉流失败 source: %s sink: %s", source, sink.String())
_ = hls.SinkManager.Remove(sink.GetID())
w.WriteHeader(http.StatusForbidden)
return
}
}
}
// 更新最近的M3U8文件
if sink != nil {
w.Write([]byte(sink.(*hls.M3U8Sink).GetPlaylist()))
return
}
// 首次拉流
context := r.Context()
m3u8Pipe := make(chan []byte, 1)
sink = hls.NewM3U8Sink(sid, source, func(m3u8 []byte) {
m3u8Pipe <- m3u8
}, sid)
sink.SetUrlValues(r.URL.Query())
if _, state := stream.PreparePlaySink(sink); utils.HookStateOK != state {
log.Sugar.Warnf("m3u8拉流失败 sink: %s", sink.String())
w.WriteHeader(http.StatusForbidden)
return
}
err := stream.SinkManager.Add(sink)
utils.Assert(err == nil)
select {
case m3u8 := <-m3u8Pipe:
// 应答M3U8文件
if m3u8 == nil {
playlist := sink.(*hls.M3U8Sink).GetPlaylist(nil)
if playlist == "" {
if playlist = sink.(*hls.M3U8Sink).GetPlaylist(r.Context()); playlist == "" {
log.Sugar.Warnf("hls拉流失败 未能生成有效m3u8文件 sink: %s source: %s", sink.GetID(), sink.GetSourceID())
w.WriteHeader(http.StatusInternalServerError)
sink.Close()
} else {
w.Write(m3u8)
return
}
break
case <-context.Done():
// 拉流端断开拉流
log.Sugar.Infof(stream.CreateSinkDisconnectionMessage(sink))
sink.Close()
break
}
w.Write([]byte(playlist))
}
func (api *ApiServer) onRtc(sourceId string, w http.ResponseWriter, r *http.Request) {
@@ -332,13 +315,11 @@ func (api *ApiServer) onRtc(sourceId string, w http.ResponseWriter, r *http.Requ
data, err := io.ReadAll(r.Body)
if err != nil {
log.Sugar.Errorf("rtc请求错误 err:%s remote:%s", err.Error(), r.RemoteAddr)
log.Sugar.Errorf("rtc拉流失败 err: %s remote: %s", err.Error(), r.RemoteAddr)
http.Error(w, err.Error(), http.StatusBadRequest)
return
} else if err := json.Unmarshal(data, &v); err != nil {
log.Sugar.Errorf("rtc请求错误 err:%s remote:%s", err.Error(), r.RemoteAddr)
log.Sugar.Errorf("rtc拉流失败 err: %s remote: %s", err.Error(), r.RemoteAddr)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@@ -365,13 +346,11 @@ func (api *ApiServer) onRtc(sourceId string, w http.ResponseWriter, r *http.Requ
group.Done()
})
sink.SetUrlValues(r.URL.Query())
log.Sugar.Infof("rtc 请求 sink:%s sdp:%v", sink.String(), v.SDP)
_, state := stream.PreparePlaySink(sink)
if utils.HookStateOK != state {
log.Sugar.Warnf("rtc 播放失败 sink:%s", sink.String())
log.Sugar.Infof("rtc拉流请求 source: %s sink: %s sdp:%v", sourceId, sink.String(), v.SDP)
ok := stream.SubscribeStream(sink, r.URL.Query())
if utils.HookStateOK != ok {
log.Sugar.Warnf("rtc拉流失败 source: %s sink: %s", sourceId, sink.String())
w.WriteHeader(http.StatusForbidden)
group.Done()
}
@@ -404,7 +383,7 @@ func (api *ApiServer) OnSourceList(w http.ResponseWriter, r *http.Request) {
ID: source.GetID(),
Protocol: source.GetType().String(),
Time: source.CreateTime(),
SinkCount: source.SinkCount(),
SinkCount: source.GetTransStreamPublisher().SinkCount(),
Bitrate: strconv.Itoa(source.GetBitrateStatistics().PreviousSecond()/1024) + "KBS", // 后续开发
Tracks: codecs,
Urls: stream.GetStreamPlayUrls(source.GetID()),
@@ -430,11 +409,11 @@ func (api *ApiServer) OnSinkList(v *IDS, w http.ResponseWriter, r *http.Request)
}
var details []SinkDetails
sinks := source.Sinks()
sinks := source.GetTransStreamPublisher().Sinks()
for _, sink := range sinks {
details = append(details,
SinkDetails{
ID: stream.SinkId2String(sink.GetID()),
ID: stream.SinkID2String(sink.GetID()),
Protocol: sink.GetProtocol().String(),
Time: sink.CreateTime(),
},
@@ -468,8 +447,12 @@ func (api *ApiServer) OnSinkClose(v *IDS, w http.ResponseWriter, r *http.Request
}
if source := stream.SourceManager.Find(v.Source); source != nil {
if sink := source.FindSink(sinkId); sink != nil {
if sink := source.GetTransStreamPublisher().FindSink(sinkId); sink != nil {
sink.Close()
if sink.GetProtocol() == stream.TransStreamHls {
_ = hls.SinkManager.Remove(sinkId)
}
}
} else {
log.Sugar.Warnf("Source with ID %s does not exist.", v.Source)

View File

@@ -35,7 +35,8 @@ type SourceSDP struct {
type GBOffer struct {
SourceSDP
AnswerSetup string `json:"answer_setup,omitempty"` // 希望应答的连接方式
AnswerSetup string `json:"answer_setup,omitempty"` // 希望应答的连接方式
TransStreamProtocol stream.TransStreamProtocol `json:"trans_stream_protocol,omitempty"`
}
func (api *ApiServer) OnGBSourceCreate(v *SourceSDP, w http.ResponseWriter, r *http.Request) {
@@ -152,71 +153,21 @@ func (api *ApiServer) OnGBOfferCreate(v *SourceSDP, w http.ResponseWriter, r *ht
}
}
func (api *ApiServer) OnGBAnswerCreate(v *GBOffer, w http.ResponseWriter, r *http.Request) {
log.Sugar.Infof("创建应答 offer: %v", v)
var sink stream.Sink
var err error
// 响应错误消息
defer func() {
if err != nil {
log.Sugar.Errorf("创建应答失败 err: %s", err.Error())
httpResponseError(w, err.Error())
if sink != nil {
sink.Close()
}
}
}()
source := stream.SourceManager.Find(v.Source)
if source == nil {
err = fmt.Errorf("%s 源不存在", v.Source)
return
}
addr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
sinkId := stream.NetAddr2SinkId(addr)
// sinkId添加随机数
if ipv4, ok := sinkId.(uint64); ok {
random := uint64(utils.RandomIntInRange(0x1000, 0xFFFF0000))
sinkId = (ipv4 & 0xFFFFFFFF00000000) | (random << 16) | (ipv4 & 0xFFFF)
}
setup := gb28181.SetupTypeFromString(v.Setup)
if v.AnswerSetup != "" {
setup = gb28181.SetupTypeFromString(v.AnswerSetup)
}
var protocol stream.TransStreamProtocol
// 级联转发
if v.SessionName == "" || v.SessionName == InviteTypePlay ||
v.SessionName == InviteTypePlayback ||
v.SessionName == InviteTypeDownload {
protocol = stream.TransStreamGBCascadedForward
} else {
// 对讲广播转发
protocol = stream.TransStreamGBTalkForward
}
func (api *ApiServer) AddForwardSink(protocol stream.TransStreamProtocol, transport stream.TransportType, sourceId string, remoteAddr string, w http.ResponseWriter, r *http.Request) {
var port int
sink, port, err = stream.NewForwardSink(setup.TransportType(), protocol, sinkId, v.Source, v.Addr, gb28181.TransportManger)
sink, port, err := stream.ForwardStream(protocol, transport, sourceId, r.URL.Query(), remoteAddr, gb28181.TransportManger)
if err != nil {
log.Sugar.Errorf("创建转发sink失败 err: %s", err.Error())
httpResponseError(w, err.Error())
return
}
log.Sugar.Infof("创建转发sink成功, sink: %s port: %d transport: %s", sink.GetID(), port, setup.TransportType())
_, state := stream.PreparePlaySink(sink)
if utils.HookStateOK != state {
err = fmt.Errorf("failed to prepare play sink")
return
}
log.Sugar.Infof("创建转发sink成功, sink: %s port: %d transport: %s", sink.GetID(), port, transport)
response := struct {
Sink string `json:"sink"` //sink id
Sink string `json:"sink"` // sink id
SDP
}{Sink: stream.SinkId2String(sinkId), SDP: SDP{Addr: net.JoinHostPort(stream.AppConfig.PublicIP, strconv.Itoa(port))}}
}{Sink: stream.SinkID2String(sink.GetID()), SDP: SDP{Addr: net.JoinHostPort(stream.AppConfig.PublicIP, strconv.Itoa(port))}}
httpResponseOK(w, &response)
}
@@ -271,3 +222,18 @@ func (api *ApiServer) OnGBTalk(w http.ResponseWriter, r *http.Request) {
talkSource.Close()
}
func (api *ApiServer) OnSinkAdd(v *GBOffer, w http.ResponseWriter, r *http.Request) {
log.Sugar.Infof("添加sink: %v", *v)
if stream.TransStreamGBCascaded != v.TransStreamProtocol && stream.TransStreamGBTalk != v.TransStreamProtocol && stream.TransStreamGBGateway != v.TransStreamProtocol {
httpResponseError(w, "不支持的协议")
return
}
setup := gb28181.SetupTypeFromString(v.Setup)
if v.AnswerSetup != "" {
setup = gb28181.SetupTypeFromString(v.AnswerSetup)
}
api.AddForwardSink(v.TransStreamProtocol, setup.TransportType(), v.Source, v.Addr, w, r)
}

View File

@@ -10,16 +10,13 @@ import (
// 处理不同包不能相互引用的需求
func NewStreamEndInfo(source stream.Source) *stream.StreamEndInfo {
tracks := source.OriginTracks()
streams := source.GetTransStreams()
func NewStreamEndInfo(source string, tracks []*stream.Track, streams map[stream.TransStreamID]stream.TransStream) *stream.StreamEndInfo {
if len(tracks) < 1 || len(streams) < 1 {
return nil
}
info := stream.StreamEndInfo{
ID: source.GetID(),
ID: source,
Timestamps: make(map[utils.AVCodecID][2]int64, len(tracks)),
}
@@ -36,7 +33,7 @@ func NewStreamEndInfo(source stream.Source) *stream.StreamEndInfo {
if stream.TransStreamHls == transStream.GetProtocol() {
if hls := transStream.(*hls.TransStream); hls.M3U8Writer.Size() > 0 {
info.M3U8Writer = hls.M3U8Writer
info.PlaylistFormat = hls.PlaylistFormat
info.PlaylistFormat = hls.PlaylistFormatPtr
}
} else if stream.TransStreamRtsp == transStream.GetProtocol() {
if rtsp := transStream.(*rtsp.TransStream); len(rtsp.Tracks) > 0 {

View File

@@ -3,6 +3,7 @@ package flv
import (
"encoding/binary"
"fmt"
"github.com/lkmio/flv"
)
const (
@@ -32,6 +33,38 @@ func GetFLVTag(block []byte) []byte {
return block[offset : length-2]
}
type TagPacket struct {
flv.Tag
Raw []byte
Offset int
}
// SplitHttpFlvBlock http-flv块分割为多个flv tag
func SplitHttpFlvBlock(httpFlv []byte) []TagPacket {
data := GetFLVTag(httpFlv)
length := len(data)
start := len(httpFlv) - length - 2
var packets []TagPacket
for i := 0; i < length; {
tag := flv.UnmarshalTag(data[i:])
offset := i
i += flv.TagHeaderSize + tag.DataSize
// 目前只需要保留第一个和最后一个tag
if offset == 0 || i >= length {
packets = append(packets, TagPacket{
Tag: tag,
Raw: data[offset:i],
Offset: start + offset,
})
}
}
return packets
}
// 计算头部的无效数据, 返回http-flv的其实位置
func computeSkipBytesSize(data []byte) int {
return int(6 + binary.BigEndian.Uint16(data[4:]))

View File

@@ -1,6 +1,7 @@
package flv
import (
"encoding/binary"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/lkm/stream"
"github.com/lkmio/transport"
@@ -18,13 +19,42 @@ func (s *Sink) StopStreaming(stream stream.TransStream) {
}
func (s *Sink) Write(index int, data []*collections.ReferenceCounter[[]byte], ts int64, keyVideo bool) error {
// 恢复推流时, 不发送9个字节的flv header
if s.prevTagSize > 0 {
data = data[1:]
s.prevTagSize = 0
var offset int
if s.SentPacketCount == 0 {
// 恢复推流时, 不发送9个字节的flv header
if s.prevTagSize > 0 {
if s.prevTagSize > 0 {
data = data[1:]
}
} else {
offset++
}
}
return s.BaseSink.Write(index, data, ts, keyVideo)
var modifiedTags []*collections.ReferenceCounter[[]byte]
// 修改第一个tag的prevTagSize, 如果第一个tag的prevTagSize与sink的prevTagSize不一致
if s.SentPacketCount < 2 {
tags := SplitHttpFlvBlock(data[offset].Get())
if s.prevTagSize != tags[0].PrevTagSize {
for _, datum := range data {
modifiedTags = append(modifiedTags, datum)
}
bytes := make([]byte, len(data[offset].Get()))
copy(bytes, data[offset].Get())
binary.BigEndian.PutUint32(bytes[tags[0].Offset:], s.prevTagSize)
modifiedTags[offset] = collections.NewReferenceCounter(bytes)
}
s.prevTagSize = uint32(11 + tags[len(tags)-1].DataSize)
}
if modifiedTags == nil {
modifiedTags = data
}
return s.BaseSink.Write(index, modifiedTags, ts, keyVideo)
}
func NewFLVSink(id stream.SinkID, sourceId string, conn net.Conn) stream.Sink {

View File

@@ -1,7 +1,6 @@
package flv
import (
"encoding/binary"
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/avformat/utils"
@@ -14,10 +13,9 @@ import (
type TransStream struct {
stream.TCPTransStream
Muxer *flv.Muxer
flvHeaderBlock []byte // 单独保存9个字节长的flv头, 只发一次, 后续恢复推流不再发送
flvExtraDataBlock []byte // metadata和sequence header
flvExtraDataPreTagSize uint32
Muxer *flv.Muxer
flvHeaderBlock []byte // 单独保存9个字节长的flv头, 只发一次, 后续恢复推流不再发送
flvExtraDataBlock []byte // metadata和sequence header
}
func (t *TransStream) Input(packet *avformat.AVPacket) ([]*collections.ReferenceCounter[[]byte], int64, bool, error) {
@@ -119,8 +117,6 @@ func (t *TransStream) WriteHeader() error {
copy(t.flvHeaderBlock[HttpFlvBlockHeaderSize:], header[:9])
copy(t.flvExtraDataBlock[HttpFlvBlockHeaderSize:], tags)
t.flvExtraDataPreTagSize = t.Muxer.PrevTagSize()
// +2 加上末尾换行符
t.flvExtraDataBlock = t.flvExtraDataBlock[:HttpFlvBlockHeaderSize+size-9+2]
writeSeparator(t.flvHeaderBlock)
@@ -139,12 +135,6 @@ func (t *TransStream) ReadKeyFrameBuffer() ([]*collections.ReferenceCounter[[]by
// 发送当前内存池已有的合并写切片
t.MWBuffer.ReadSegmentsFromKeyFrameIndex(func(segment *collections.ReferenceCounter[[]byte]) {
// 修改第一个flv tag的pre tag size为sequence header tag size
bytes := segment.Get()
if t.OutBufferSize < 1 {
binary.BigEndian.PutUint32(GetFLVTag(bytes), t.flvExtraDataPreTagSize)
}
t.AppendOutStreamBuffer(segment)
})
@@ -185,7 +175,7 @@ func TransStreamFactory(source stream.Source, protocol stream.TransStreamProtoco
var prevTagSize uint32
var metaData *amf0.Object
endInfo := source.GetStreamEndInfo()
endInfo := source.GetTransStreamPublisher().GetStreamEndInfo()
if endInfo != nil {
prevTagSize = endInfo.FLVPrevTagSize
}

View File

@@ -1,9 +0,0 @@
package gb28181
import (
"github.com/lkmio/lkm/stream"
)
func CascadedTransStreamFactory(source stream.Source, protocol stream.TransStreamProtocol, tracks []*stream.Track) (stream.TransStream, error) {
return stream.NewRtpTransStream(stream.TransStreamGBCascadedForward, 1024), nil
}

100
gb28181/gateway.go Normal file
View File

@@ -0,0 +1,100 @@
package gb28181
import (
"encoding/binary"
"fmt"
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/log"
"github.com/lkmio/lkm/stream"
"github.com/lkmio/mpeg"
"github.com/lkmio/rtp"
)
type GBGateway struct {
stream.BaseTransStream
ps *mpeg.PSMuxer
psBuffer []byte
tracks map[utils.AVCodecID]struct {
index int
rtp rtp.Muxer
}
}
func (s *GBGateway) WriteHeader() error {
if len(s.tracks) == 0 {
return fmt.Errorf("no tracks available")
}
return nil
}
func (s *GBGateway) AddTrack(track *stream.Track) error {
s.BaseTransStream.AddTrack(track)
var muxer rtp.Muxer
if utils.AVCodecIdH264 == track.Stream.CodecID || utils.AVCodecIdH265 == track.Stream.CodecID || utils.AVCodecIdAAC == track.Stream.CodecID || utils.AVCodecIdPCMALAW == track.Stream.CodecID || utils.AVCodecIdPCMMULAW == track.Stream.CodecID {
muxer = rtp.NewMuxer(96, 0, 0xFFFFFFFF)
} else {
log.Sugar.Errorf("不支持的编码格式: %d", track.Stream.CodecID)
return nil
}
index, err := s.ps.AddTrack(track.Stream.MediaType, track.Stream.CodecID)
if err != nil {
log.Sugar.Error("添加%s到ps muxer失败", track.Stream.CodecID)
return nil
}
s.tracks[track.Stream.CodecID] = struct {
index int
rtp rtp.Muxer
}{index: index, rtp: muxer}
return nil
}
func (s *GBGateway) Input(packet *avformat.AVPacket) ([]*collections.ReferenceCounter[[]byte], int64, bool, error) {
track, ok := s.tracks[packet.CodecID]
if !ok {
log.Sugar.Errorf("未找到对应的track: %d", packet.CodecID)
return nil, 0, false, nil
}
dts := packet.ConvertDts(90000)
pts := packet.ConvertPts(90000)
data := avformat.AVCCPacket2AnnexB(s.BaseTransStream.Tracks[packet.Index].Stream, packet)
if cap(s.psBuffer) < len(data)+1024*64 {
s.psBuffer = make([]byte, len(data)*2)
}
n := s.ps.Input(s.psBuffer, track.index, packet.Key, data, &pts, &dts)
var result []*collections.ReferenceCounter[[]byte]
var rtpBuffer []byte
track.rtp.Input(s.psBuffer[:n], uint32(dts), func() []byte {
rtpBuffer = stream.UDPReceiveBufferPool.Get().([]byte)
return rtpBuffer[2:]
}, func(bytes []byte) {
binary.BigEndian.PutUint16(rtpBuffer, uint16(len(bytes)))
refPacket := collections.NewReferenceCounter(rtpBuffer[:2+len(bytes)])
result = append(result, refPacket)
})
return result, 0, true, nil
}
func NewGBGateway() *GBGateway {
return &GBGateway{
ps: mpeg.NewPsMuxer(),
psBuffer: make([]byte, 1024*1024*2),
tracks: make(map[utils.AVCodecID]struct {
index int
rtp rtp.Muxer
}),
}
}
func GatewayTransStreamFactory(source stream.Source, protocol stream.TransStreamProtocol, tracks []*stream.Track) (stream.TransStream, error) {
return NewGBGateway(), nil
}

View File

@@ -1,6 +1,7 @@
package gb28181
import (
"encoding/binary"
"fmt"
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/utils"
@@ -97,6 +98,8 @@ type BaseGBSource struct {
audioPacketCreatedTime int64
videoPacketCreatedTime int64
isSystemClock bool // 推流时间戳不正确, 是否使用系统时间.
lastRtpTimestamp int64
sameTimePackets [][]byte
}
func (source *BaseGBSource) Init(receiveQueueSize int) {
@@ -108,19 +111,40 @@ func (source *BaseGBSource) Init(receiveQueueSize int) {
source.SetType(stream.SourceType28181)
source.probeBuffer = mpeg.NewProbeBuffer(PsProbeBufferSize)
source.PublishSource.Init(receiveQueueSize)
source.lastRtpTimestamp = -1
}
// Input 输入rtp包, 处理PS流, 负责解析->封装->推流
func (source *BaseGBSource) Input(data []byte) error {
// 国标级联转发
if source.ForwardTransStream != nil {
packet := avformat.AVPacket{Data: data}
source.DispatchPacket(source.ForwardTransStream, &packet)
}
packet := rtp.Packet{}
_ = packet.Unmarshal(data)
// 国标级联转发
if source.GetTransStreamPublisher().GetTransStreams() != nil {
if source.lastRtpTimestamp == -1 {
source.lastRtpTimestamp = int64(packet.Timestamp)
}
// 相同时间戳的RTP包, 积攒一起发送, 降低管道压力
length := len(data)
if int64(packet.Timestamp) != source.lastRtpTimestamp {
source.lastRtpTimestamp = int64(packet.Timestamp)
if len(source.sameTimePackets) > 0 {
source.GetTransStreamPublisher().Post(&stream.StreamEvent{Type: stream.StreamEventTypeRawPacket, Data: source.sameTimePackets})
source.sameTimePackets = nil
}
}
if stream.UDPReceiveBufferSize-2 < length {
log.Sugar.Errorf("rtp包过大, 不转发. source: %s ssrc: %x size: %d", source.ID, source.ssrc, len(data))
} else {
bytes := stream.UDPReceiveBufferPool.Get().([]byte)
copy(bytes[2:], data)
binary.BigEndian.PutUint16(bytes[:2], uint16(length))
source.sameTimePackets = append(source.sameTimePackets, bytes[:2+length])
}
}
var bytes []byte
var n int
var err error

View File

@@ -31,7 +31,7 @@ func (s *TalkStream) Input(packet *avformat.AVPacket) ([]*collections.ReferenceC
func NewTalkTransStream() (stream.TransStream, error) {
return &TalkStream{
RtpStream: stream.NewRtpTransStream(stream.TransStreamGBTalkForward, 1024),
RtpStream: stream.NewRtpTransStream(stream.TransStreamGBTalk, 1024),
muxer: rtp.NewMuxer(8, 0, 0xFFFFFFFF),
packet: make([]byte, 1500),
}, nil

View File

@@ -36,6 +36,9 @@ func DecodeGBRTPOverTCPPacket(data []byte, source GBSource, decoder *transport.L
}
i += n
if bytes == nil {
break
}
// 单端口模式,ssrc匹配source
if source == nil || stream.SessionStateHandshakeSuccess == source.State() {

View File

@@ -1,11 +1,12 @@
package hls
import (
"context"
"fmt"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/log"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/lkm/stream"
"strings"
"sync/atomic"
"time"
)
@@ -15,63 +16,40 @@ const (
type M3U8Sink struct {
stream.BaseSink
cb func(m3u8 []byte) // 生成m3u8文件的发送回调
sessionId string // 拉流会话ID
playtime time.Time
playTimer *time.Timer
playlistFormat *string
playingTime atomic.Value
sessionId string // 拉流会话ID
playlistFormat *string
m3u8ReadyCtx context.Context
m3u8ReadyCancel func()
}
// SendM3U8Data 首次向拉流端应答M3U8文件 后续更新M3U8文件, 通过调用@see GetPlaylist 函数获取最新的M3U8文件.
func (s *M3U8Sink) SendM3U8Data(data *string) error {
utils.Assert(data != nil)
utils.Assert(s.playlistFormat == nil)
s.playlistFormat = data
s.cb([]byte(s.GetPlaylist()))
// 开启计时器, 长时间没有拉流关闭sink
timeout := time.Duration(stream.AppConfig.IdleTimeout)
if timeout < time.Second {
timeout = time.Duration(stream.AppConfig.Hls.Duration) * 2 * 3 * time.Second
}
s.playTimer = time.AfterFunc(timeout, func() {
sub := time.Now().Sub(s.playtime)
if sub > timeout {
log.Sugar.Errorf("hls拉流超时 sink: %s ", s.ID)
s.Close()
return
}
s.playTimer.Reset(timeout)
})
return nil
}
func (s *M3U8Sink) StartStreaming(transStream stream.TransStream) error {
if s.playlistFormat != nil {
return nil
}
hls := transStream.(*TransStream)
if hls.M3U8Writer.Size() > 0 && s.playlistFormat == nil {
if err := s.SendM3U8Data(hls.PlaylistFormat); err != nil {
return err
}
} else {
// m3u8文件中还没有切片时, 将sink添加到等待队列
hls.m3u8Sinks[s.GetID()] = s
func (s *M3U8Sink) Write(index int, data []*collections.ReferenceCounter[[]byte], ts int64, keyVideo bool) error {
if s.playlistFormat == nil {
s.playlistFormat = bytesToStringPtr(data[0].Get())
s.m3u8ReadyCancel()
}
return nil
}
func (s *M3U8Sink) GetPlaylist() string {
func (s *M3U8Sink) GetPlaylist(ctx context.Context) string {
// 更新拉流时间
//s.RefreshPlayTime()
s.RefreshPlayingTime()
if s.playlistFormat == nil {
if ctx == nil {
return ""
}
select {
case <-ctx.Done():
return ""
case <-s.m3u8ReadyCtx.Done():
if s.playlistFormat == nil {
return ""
}
}
}
// 替换每个sink唯一的拉流会话ID
param := fmt.Sprintf("?%s=%s", SessionIDKey, s.sessionId)
@@ -79,24 +57,30 @@ func (s *M3U8Sink) GetPlaylist() string {
return playlist
}
func (s *M3U8Sink) RefreshPlayTime() {
s.playtime = time.Now()
func (s *M3U8Sink) RefreshPlayingTime() {
s.playingTime.Store(time.Now())
}
func (s *M3U8Sink) GetPlayingTime() time.Time {
if t := s.playingTime.Load(); t != nil {
return t.(time.Time)
}
return time.Time{}
}
func (s *M3U8Sink) Close() {
s.m3u8ReadyCancel()
s.BaseSink.Close()
stream.SinkManager.Remove(s.ID)
if s.playTimer != nil {
s.playTimer.Stop()
s.playTimer = nil
}
SinkManager.Remove(s.ID)
}
func NewM3U8Sink(id stream.SinkID, sourceId string, cb func(m3u8 []byte), sessionId string) stream.Sink {
func NewM3U8Sink(id stream.SinkID, sourceId string, sessionId string) stream.Sink {
ctx, cancel := context.WithCancel(context.Background())
return &M3U8Sink{
BaseSink: stream.BaseSink{ID: id, SourceID: sourceId, Protocol: stream.TransStreamHls, TCPStreaming: true},
cb: cb,
sessionId: sessionId,
BaseSink: stream.BaseSink{ID: id, SourceID: sourceId, Protocol: stream.TransStreamHls, TCPStreaming: true},
sessionId: sessionId,
m3u8ReadyCtx: ctx,
m3u8ReadyCancel: cancel,
}
}

View File

@@ -11,6 +11,7 @@ import (
"os"
"path/filepath"
"strconv"
"unsafe"
)
type TransStream struct {
@@ -34,13 +35,14 @@ type TransStream struct {
duration int // 切片时长, 单位秒
playlistLength int // 最大切片文件个数
m3u8Sinks map[stream.SinkID]*M3U8Sink // 保存还未生成mu38文件就拉流的sink, 发送一次后就删除.
PlaylistFormat *string // 位于内存中的播放列表每个sink都引用指针地址.
PlaylistFormatPtr *string // 位于内存中的m3u8播放列表每个sink都引用指针地址.
PlaylistFormatPtrCounter []*collections.ReferenceCounter[[]byte] // string指针转byte[], 方便发送给sink
}
func (t *TransStream) Input(packet *avformat.AVPacket) ([]*collections.ReferenceCounter[[]byte], int64, bool, error) {
// 创建一下个切片
// 已缓存时长>=指定时长, 如果存在视频, 还需要等遇到关键帧才切片
var newSegment bool
if (!t.ExistVideo || utils.AVMediaTypeVideo == packet.MediaType && packet.Key) && float32(t.muxer.Duration())/90000 >= float32(t.duration) {
// 保存当前切片文件
if t.ctx.file != nil {
@@ -54,6 +56,8 @@ func (t *TransStream) Input(packet *avformat.AVPacket) ([]*collections.Reference
if err := t.createSegment(); err != nil {
return nil, -1, false, err
}
newSegment = true
}
pts := packet.ConvertPts(90000)
@@ -63,6 +67,7 @@ func (t *TransStream) Input(packet *avformat.AVPacket) ([]*collections.Reference
data = avformat.AVCCPacket2AnnexB(t.BaseTransStream.Tracks[packet.Index].Stream, packet)
}
// 写入ts切片
length := len(data)
capacity := cap(t.ctx.writeBuffer)
for i := 0; i < length; {
@@ -75,6 +80,12 @@ func (t *TransStream) Input(packet *avformat.AVPacket) ([]*collections.Reference
i += t.muxer.Input(bytes, packet.Index, data[i:], length, dts, pts, packet.Key, i == 0)
t.ctx.writeBufferSize += mpeg.TsPacketSize
}
// 缓存完第二个切片, 才响应发送m3u8文件. 如果一个切片就发, 播放器缓存少会卡顿.
if newSegment && t.M3U8Writer.Size() > 1 {
return t.PlaylistFormatPtrCounter, -1, true, nil
}
return nil, -1, true, nil
}
@@ -126,7 +137,7 @@ func (t *TransStream) flushSegment(end bool) error {
// m3u8Txt += "#EXT-X-ENDLIST"
//}
*t.PlaylistFormat = m3u8Txt
*t.PlaylistFormatPtr = m3u8Txt
// 写入最新的m3u8到文件
if t.m3u8File != nil {
@@ -139,16 +150,6 @@ func (t *TransStream) flushSegment(end bool) error {
}
}
// 通知等待m3u8的sink
// 缓存完第二个切片, 才响应发送m3u8文件. 如果一个切片就发, 播放器缓存少会卡顿.
if len(t.m3u8Sinks) > 0 && t.M3U8Writer.Size() > 1 {
for _, sink := range t.m3u8Sinks {
sink.SendM3U8Data(t.PlaylistFormat)
}
t.m3u8Sinks = make(map[stream.SinkID]*M3U8Sink, 0)
}
return nil
}
@@ -210,16 +211,19 @@ func (t *TransStream) Close() ([]*collections.ReferenceCounter[[]byte], int64, e
t.m3u8File = nil
}
// 如果关闭HLS输出流时, 没有有效切片(推流数据过少), 通知等待的sink
for _, sink := range t.m3u8Sinks {
sink.cb(nil)
}
t.m3u8Sinks = nil
return nil, 0, err
}
func stringPtrToBytes(ptr *string) []byte {
ptrAddr := uintptr(unsafe.Pointer(ptr))
return (*[unsafe.Sizeof(ptr)]byte)(unsafe.Pointer(&ptrAddr))[:]
}
func bytesToStringPtr(b []byte) *string {
ptrAddr := *(*uintptr)(unsafe.Pointer(&b[0]))
return (*string)(unsafe.Pointer(ptrAddr))
}
func DeleteOldSegments(id string) {
var index int
for ; ; index++ {
@@ -274,11 +278,13 @@ func NewTransStream(dir, m3u8Name, tsFormat, tsUrl string, segmentDuration, play
}
if playlistFormat != nil {
transStream.PlaylistFormat = playlistFormat
transStream.PlaylistFormatPtr = playlistFormat
} else {
transStream.PlaylistFormat = new(string)
transStream.PlaylistFormatPtr = new(string)
}
playlistFormatPtrCounter := collections.NewReferenceCounter[[]byte](stringPtrToBytes(transStream.PlaylistFormatPtr))
transStream.PlaylistFormatPtrCounter = append(transStream.PlaylistFormatPtrCounter, playlistFormatPtrCounter)
// 创建TS封装器
muxer := mpeg.NewTSMuxer()
@@ -288,7 +294,6 @@ func NewTransStream(dir, m3u8Name, tsFormat, tsUrl string, segmentDuration, play
transStream.muxer = muxer
transStream.m3u8File = file
transStream.m3u8Sinks = make(map[stream.SinkID]*M3U8Sink, 24)
return transStream, nil
}
@@ -299,7 +304,7 @@ func TransStreamFactory(source stream.Source, protocol stream.TransStreamProtoco
var playlistFormat *string
startSeq := -1
endInfo := source.GetStreamEndInfo()
endInfo := source.GetTransStreamPublisher().GetStreamEndInfo()
if endInfo != nil && endInfo.M3U8Writer != nil {
writer = endInfo.M3U8Writer
playlistFormat = endInfo.PlaylistFormat

67
hls/sink_manager.go Normal file
View File

@@ -0,0 +1,67 @@
package hls
import (
"github.com/lkmio/lkm/log"
"github.com/lkmio/lkm/stream"
"sync"
"time"
)
// SinkManager 目前只用于保存HLS拉流Sink
var SinkManager *sinkManager
func init() {
SinkManager = &sinkManager{}
}
type sinkManager struct {
m sync.Map
}
func (s *sinkManager) Add(sink stream.Sink) bool {
_, ok := s.m.LoadOrStore(sink.GetID(), sink)
return !ok
}
func (s *sinkManager) Find(id stream.SinkID) stream.Sink {
value, ok := s.m.Load(id)
if ok {
return value.(stream.Sink)
}
return nil
}
func (s *sinkManager) Remove(id stream.SinkID) stream.Sink {
value, loaded := s.m.LoadAndDelete(id)
if loaded {
return value.(stream.Sink)
}
return nil
}
func (s *sinkManager) Exist(id stream.SinkID) bool {
_, ok := s.m.Load(id)
return ok
}
// StartPullStreamTimeoutTimer 开启hls拉流计时器, 用于检测hls拉流超时
func (s *sinkManager) StartPullStreamTimeoutTimer(interval, timeout time.Duration) {
var timer *time.Timer
timer = time.AfterFunc(interval, func() {
now := time.Now()
s.m.Range(func(key, value any) bool {
sink := value.(*M3U8Sink)
playingTime := sink.GetPlayingTime()
if now.Sub(playingTime) > timeout {
log.Sugar.Infof("hls拉流超时 sink: %s playingTime: %s", sink.GetID(), playingTime.Format("2006-01-02 15:04:05"))
go sink.Close()
}
return true
})
timer.Reset(interval)
})
}

View File

@@ -4,6 +4,7 @@ import (
"encoding/binary"
"fmt"
"github.com/lkmio/avformat/utils"
"strings"
)
const (
@@ -27,6 +28,7 @@ const (
PTAudioADPCMA = 26
)
// Packet 1078-2016音视频和透传包. 视频帧包头26个字节, 音频帧22个字节, 透传数据14个字节.
type Packet struct {
pt byte
packetType byte
@@ -38,19 +40,26 @@ type Packet struct {
}
func (p *Packet) Unmarshal(data []byte) error {
if len(data) < 12 {
length := len(data)
if length < 12 {
return fmt.Errorf("invaild data")
}
packetType := data[11] >> 4 & 0x0F
// 忽略透传数据
if TransmissionDataMark == packetType {
return fmt.Errorf("invaild data")
}
// 忽略低于最低长度的数据包
if (AudioFrameMark == packetType && len(data) < 26) || (AudioFrameMark == packetType && len(data) < 22) {
return fmt.Errorf("invaild data")
if packetType < AudioFrameMark {
if length < 26 {
return fmt.Errorf("invaild data")
}
} else if AudioFrameMark == packetType {
if length < 22 {
return fmt.Errorf("invaild data")
}
} else if TransmissionDataMark == packetType {
if length < 14 {
return fmt.Errorf("invaild data")
}
} else {
return fmt.Errorf("unknown packet type %x", packetType)
}
// x扩展位,固定为0
@@ -64,6 +73,7 @@ func (p *Packet) Unmarshal(data []byte) error {
simNumber += fmt.Sprintf("%02x", data[i])
}
simNumber = strings.TrimLeft(simNumber, "0")
// channel
channelNumber := data[10]
// subMark
@@ -71,11 +81,13 @@ func (p *Packet) Unmarshal(data []byte) error {
// 时间戳,单位ms
var ts uint64
n := 12
// 音视频帧才有时间戳字段
if TransmissionDataMark != packetType {
ts = binary.BigEndian.Uint64(data[n:])
n += 8
}
// 与上一帧的间隔时间戳, 视频帧才有此字段
if AudioFrameMark > packetType {
// iFrameInterval
_ = binary.BigEndian.Uint16(data[n:])

View File

@@ -2,17 +2,114 @@ package jt1078
import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/gorilla/mux"
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/bufio"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/gb28181"
"github.com/lkmio/lkm/stream"
"github.com/lkmio/mpeg"
"github.com/lkmio/transport"
"net"
"net/http"
"os"
"testing"
"time"
)
func TestPublish(t *testing.T) {
type Handler struct {
muxer *mpeg.PSMuxer
fos *os.File
buffer []byte
tracks map[int]int
gateway *gb28181.GBGateway
udp *transport.UDPClient
}
func (h Handler) OnNewTrack(track avformat.Track) {
addTrack, err := h.muxer.AddTrack(track.GetStream().MediaType, track.GetStream().CodecID)
if err != nil {
println(err.Error())
} else {
h.tracks[track.GetStream().Index] = addTrack
h.gateway.AddTrack(&stream.Track{Stream: track.GetStream()})
}
}
func (h Handler) OnTrackComplete() {
}
func (h Handler) OnTrackNotFind() {
//TODO implement me
panic("implement me")
}
func (h Handler) OnPacket(packet *avformat.AVPacket) {
i, ok := h.tracks[packet.Index]
if !ok {
return
}
dts := packet.ConvertDts(90000)
pts := packet.ConvertPts(90000)
var n int
if packet.MediaType == utils.AVMediaTypeVideo {
// 1078流已经是annexb打包
// annexBData := avformat.AVCCPacket2AnnexB(t.BaseTransStream.Tracks[packet.Index].Stream, packet)
n = h.muxer.Input(h.buffer, i, packet.Key, packet.Data, &pts, &dts)
} else {
n = h.muxer.Input(h.buffer, i, true, packet.Data, &pts, &dts)
}
if n > 0 {
h.fos.Write(h.buffer[:n])
}
packets, _, _, err := h.gateway.Input(packet)
if err != nil {
panic(err)
}
for _, refPacket := range packets {
bytes := refPacket.Get()
err = h.udp.Write(bytes[2:])
if err != nil {
panic(err)
}
}
}
func publish() {
//path := "../../source_files/10352264314-2.bin"
path := "../../source_files/013800138000-1.bin"
client := transport.TCPClient{}
addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:1078")
if err != nil {
panic(err)
}
_, err = client.Connect(nil, addr)
if err != nil {
panic(err)
}
file, err := os.ReadFile(path)
if err != nil {
panic(err)
}
index := 0
for index < len(file) {
n := bufio.MinInt(len(file)-index, 1500)
client.Write(file[index : index+n])
index += n
time.Sleep(1 * time.Millisecond)
}
}
func TestPublish(t *testing.T) {
t.Run("decode_1078_data", func(t *testing.T) {
data, err := os.ReadFile("../dump/jt1078-127.0.0.1.50659")
if err != nil {
@@ -55,31 +152,98 @@ func TestPublish(t *testing.T) {
})
t.Run("publish", func(t *testing.T) {
publish()
})
// 1078->ps->rtp
// 1078封装成ps流保存到文件, 再用rtp打包发送出去, 用wireshark导出ps流看播放是否正常
t.Run("jt2gb", func(t *testing.T) {
//path := "../../source_files/10352264314-2.bin"
path := "../../source_files/013800138000-1.bin"
client := transport.TCPClient{}
addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:1078")
if err != nil {
panic(err)
}
_, err = client.Connect(nil, addr)
if err != nil {
panic(err)
}
file, err := os.ReadFile(path)
if err != nil {
panic(err)
}
index := 0
for index < len(file) {
n := bufio.MinInt(len(file)-index, 1500)
client.Write(file[index : index+n])
index += n
time.Sleep(1 * time.Millisecond)
openFile, err := os.OpenFile(path+".ps", os.O_CREATE|os.O_RDWR, 0666)
if err != nil {
panic(err)
}
client := &transport.UDPClient{}
addr, _ := net.ResolveUDPAddr("udp", "127.0.0.1:10000")
err = client.Connect(nil, addr)
if err != nil {
panic(err)
}
demuxer := NewDemuxer()
demuxer.SetHandler(&Handler{
muxer: mpeg.NewPsMuxer(),
buffer: make([]byte, 1024*1024*2),
fos: openFile,
tracks: make(map[int]int),
gateway: gb28181.NewGBGateway(),
udp: client,
})
defer demuxer.Close()
delimiter := [4]byte{0x30, 0x31, 0x63, 0x64}
decoder := transport.NewDelimiterFrameDecoder(1024*1024*2, delimiter[:])
var n int
for {
r, bytes, err := decoder.Input(file[n:])
if err != nil || bytes == nil {
break
}
n += r
_, err = demuxer.Input(bytes)
if err != nil {
panic(err)
}
}
})
// hook gb-cms的on_invite回调, 处理invite请求, 推送本地文件,发送200响应
t.Run("hook_on_invite", func(t *testing.T) {
// 创建http server
router := mux.NewRouter()
// 示例路由
router.HandleFunc("/api/v1/jt1078/on_invite", func(w http.ResponseWriter, r *http.Request) {
v := struct {
SimNumber string `json:"sim_number,omitempty"`
ChannelNumber string `json:"channel_number,omitempty"`
}{}
// 读取请求体
bytes := make([]byte, 1024)
n, err := r.Body.Read(bytes)
if n < 1 {
panic(err)
}
err = json.Unmarshal(bytes[:n], &v)
if err != nil {
panic(err)
}
fmt.Printf("on_invite sim_number: %s, channel_number: %s\r\n", v.SimNumber, v.ChannelNumber)
w.WriteHeader(http.StatusOK)
go publish()
})
server := &http.Server{
Addr: "localhost:8081",
Handler: router,
WriteTimeout: 15 * time.Second,
ReadTimeout: 15 * time.Second,
}
err := server.ListenAndServe()
if err != nil {
panic(err)
}
})
}

11
main.go
View File

@@ -9,6 +9,7 @@ import (
"github.com/lkmio/lkm/rtsp"
"github.com/lkmio/transport"
"os"
"time"
"github.com/lkmio/lkm/gb28181"
"github.com/lkmio/lkm/log"
@@ -28,8 +29,9 @@ func init() {
stream.RegisterTransStreamFactory(stream.TransStreamFlv, flv.TransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamRtsp, rtsp.TransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamRtc, rtc.TransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamGBCascadedForward, gb28181.CascadedTransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamGBTalkForward, gb28181.TalkTransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamGBCascaded, stream.GBCascadedTransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamGBTalk, gb28181.TalkTransStreamFactory)
stream.RegisterTransStreamFactory(stream.TransStreamGBGateway, gb28181.GatewayTransStreamFactory)
stream.SetRecordStreamFactory(record.NewFLVFileSink)
stream.StreamEndInfoBride = NewStreamEndInfo
@@ -120,6 +122,11 @@ func main() {
log.Sugar.Info("启动rtsp服务成功 addr:", rtspAddr.String())
}
if stream.AppConfig.Hls.Enable {
// 每10秒检查一次hls拉流超时, 60秒内没有拉流的sink将被关闭
hls.SinkManager.StartPullStreamTimeoutTimer(10*time.Second, 60*time.Second)
}
log.Sugar.Info("启动http服务 addr:", stream.ListenAddr(stream.AppConfig.Http.Port))
go startApiServer(net.JoinHostPort(stream.AppConfig.ListenIP, strconv.Itoa(stream.AppConfig.Http.Port)))

View File

@@ -56,19 +56,18 @@ func (s *Session) OnPlay(app, stream_ string) utils.HookState {
streamName, values := stream.ParseUrl(stream_)
sourceId := s.generateSourceID(app, streamName)
sink := NewSink(stream.NetAddr2SinkId(s.conn.RemoteAddr()), sourceId, s.conn, s.stack)
sink.SetUrlValues(values)
sinkId := stream.NetAddr2SinkID(s.conn.RemoteAddr())
log.Sugar.Infof("rtmp onplay app: %s stream: %s sink: %v conn: %s", app, stream_, sinkId, s.conn.RemoteAddr().String())
log.Sugar.Infof("rtmp onplay app: %s stream: %s sink: %v conn: %s", app, stream_, sink.GetID(), s.conn.RemoteAddr().String())
_, state := stream.PreparePlaySink(sink)
if utils.HookStateOK != state {
sink := NewSink(sinkId, sourceId, s.conn, s.stack)
ok := stream.SubscribeStream(sink, values)
if utils.HookStateOK != ok {
log.Sugar.Errorf("rtmp拉流失败 source: %s sink: %s", sourceId, sink.GetID())
} else {
s.handle = sink
}
return state
return ok
}
func (s *Session) Input(data []byte) error {

View File

@@ -1,6 +1,5 @@
package rtsp
import (
"fmt"
"github.com/lkmio/avformat/utils"
@@ -125,7 +124,7 @@ func (h handler) OnDescribe(request Request) (*http.Response, []byte, error) {
}
}
sinkId := stream.NetAddr2SinkId(request.session.conn.RemoteAddr())
sinkId := stream.NetAddr2SinkID(request.session.conn.RemoteAddr())
sink := NewSink(sinkId, request.sourceId, request.session.conn, func(sdp string) {
// 响应sdp回调
response = NewOKResponse(request.headers.Get("Cseq"))
@@ -133,10 +132,9 @@ func (h handler) OnDescribe(request Request) (*http.Response, []byte, error) {
request.session.response(response, []byte(sdp))
})
sink.SetUrlValues(request.url.Query())
_, code := stream.PreparePlaySinkWithReady(sink, false)
if utils.HookStateOK != code {
return nil, nil, fmt.Errorf("hook failed. code: %d", code)
ok := stream.SubscribeStreamWithOptions(sink, request.url.Query(), false, false)
if utils.HookStateOK != ok {
return nil, nil, fmt.Errorf("hook failed. code: %d", ok)
}
request.session.sink = sink.(*Sink)
@@ -234,7 +232,7 @@ func (h handler) OnPlay(request Request) (*http.Response, []byte, error) {
return nil, nil, fmt.Errorf("Source with ID %s does not exist.", request.sourceId)
}
source.AddSink(sink)
source.GetTransStreamPublisher().AddSink(sink)
return response, nil, nil
}

View File

@@ -292,7 +292,7 @@ func NewTransStream(addr net.IPAddr, urlFormat string, oldTracks map[byte]uint16
func TransStreamFactory(source stream.Source, protocol stream.TransStreamProtocol, tracks []*stream.Track) (stream.TransStream, error) {
trackFormat := "?track=%d"
var oldTracks map[byte]uint16
if endInfo := source.GetStreamEndInfo(); endInfo != nil {
if endInfo := source.GetTransStreamPublisher().GetStreamEndInfo(); endInfo != nil {
oldTracks = endInfo.RtspTracks
}

View File

@@ -43,7 +43,7 @@ func (f *ForwardSink) OnConnected(conn net.Conn) []byte {
// 如果f.Conn赋值后, 发送数据先于EnableAsyncWriteMode执行, 可能会panic
// 所以保险一点, 放在主协程执行
ExecuteSyncEventOnSource(f.SourceID, func() {
ExecuteSyncEventOnTransStreamPublisher(f.SourceID, func() {
f.Conn = conn
f.BaseSink.EnableAsyncWriteMode(512)
})
@@ -68,7 +68,9 @@ func (f *ForwardSink) Write(index int, data []*collections.ReferenceCounter[[]by
}
if TransportTypeUDP == f.transportType {
f.socket.(*transport.UDPClient).Write(data[0].Get()[2:])
for _, datum := range data {
f.socket.(*transport.UDPClient).Write(datum.Get()[2:])
}
} else {
return f.BaseSink.Write(index, data, ts, keyVideo)
}
@@ -91,7 +93,7 @@ func (f *ForwardSink) Close() {
// StartReceiveTimer 启动tcp sever计时器, 如果计时器触发, 没有连接, 则关闭流
func (f *ForwardSink) StartReceiveTimer() {
f.receiveTimer = time.AfterFunc(time.Second*10, func() {
f.receiveTimer = time.AfterFunc(ForwardSinkWaitTimeout*time.Second, func() {
if f.Conn == nil {
log.Sugar.Infof("%s 等待连接超时, 关闭sink", f.Protocol)
f.Close()

View File

@@ -10,30 +10,30 @@ import (
type GOPBuffer interface {
// AddPacket Return bool 缓存帧是否成功, 如果首帧非关键帧, 缓存失败
AddPacket(packet *avformat.AVPacket) bool
AddPacket(packet *collections.ReferenceCounter[*avformat.AVPacket]) bool
PeekAll(handler func(packet *avformat.AVPacket))
PeekAll(handler func(*collections.ReferenceCounter[*avformat.AVPacket]))
Peek(index int) *avformat.AVPacket
Peek(index int) *collections.ReferenceCounter[*avformat.AVPacket]
PopAll(handler func(packet *avformat.AVPacket))
PopAll(handler func(*collections.ReferenceCounter[*avformat.AVPacket]))
RequiresClear(nextPacket *avformat.AVPacket) bool
RequiresClear(nextPacket *collections.ReferenceCounter[*avformat.AVPacket]) bool
Size() int
}
type streamBuffer struct {
buffer collections.RingBuffer[*avformat.AVPacket]
buffer collections.RingBuffer[*collections.ReferenceCounter[*avformat.AVPacket]]
hasVideoKeyFrame bool
}
func (s *streamBuffer) AddPacket(packet *avformat.AVPacket) bool {
if utils.AVMediaTypeVideo == packet.MediaType {
if packet.Key {
func (s *streamBuffer) AddPacket(packet *collections.ReferenceCounter[*avformat.AVPacket]) bool {
if utils.AVMediaTypeVideo == packet.Get().MediaType {
if packet.Get().Key {
s.hasVideoKeyFrame = true
} else if !s.hasVideoKeyFrame {
// 丢弃首帧视频非关键帧
// 丢弃首帧非关键视频
return false
}
}
@@ -42,7 +42,7 @@ func (s *streamBuffer) AddPacket(packet *avformat.AVPacket) bool {
return true
}
func (s *streamBuffer) Peek(index int) *avformat.AVPacket {
func (s *streamBuffer) Peek(index int) *collections.ReferenceCounter[*avformat.AVPacket] {
utils.Assert(index < s.buffer.Size())
head, tail := s.buffer.Data()
@@ -53,7 +53,7 @@ func (s *streamBuffer) Peek(index int) *avformat.AVPacket {
}
}
func (s *streamBuffer) PeekAll(handler func(packet *avformat.AVPacket)) {
func (s *streamBuffer) PeekAll(handler func(packet *collections.ReferenceCounter[*avformat.AVPacket])) {
head, tail := s.buffer.Data()
if head != nil {
@@ -73,7 +73,7 @@ func (s *streamBuffer) Size() int {
return s.buffer.Size()
}
func (s *streamBuffer) PopAll(handler func(packet *avformat.AVPacket)) {
func (s *streamBuffer) PopAll(handler func(packet *collections.ReferenceCounter[*avformat.AVPacket])) {
for s.buffer.Size() > 0 {
pkt := s.buffer.Pop()
handler(pkt)
@@ -82,10 +82,10 @@ func (s *streamBuffer) PopAll(handler func(packet *avformat.AVPacket)) {
s.hasVideoKeyFrame = false
}
func (s *streamBuffer) RequiresClear(nextPacket *avformat.AVPacket) bool {
return s.Size()+1 == s.buffer.Capacity() || (s.hasVideoKeyFrame && utils.AVMediaTypeVideo == nextPacket.MediaType && nextPacket.Key)
func (s *streamBuffer) RequiresClear(nextPacket *collections.ReferenceCounter[*avformat.AVPacket]) bool {
return s.Size()+1 == s.buffer.Capacity() || (s.hasVideoKeyFrame && utils.AVMediaTypeVideo == nextPacket.Get().MediaType && nextPacket.Get().Key)
}
func NewStreamBuffer() GOPBuffer {
return &streamBuffer{buffer: collections.NewRingBuffer[*avformat.AVPacket](1000), hasVideoKeyFrame: false}
return &streamBuffer{buffer: collections.NewRingBuffer[*collections.ReferenceCounter[*avformat.AVPacket]](1000), hasVideoKeyFrame: false}
}

View File

@@ -13,7 +13,7 @@ import (
// 每个通知事件都需要携带的字段
type eventInfo struct {
Stream string `json:"stream"` //stream GetID
Protocol string `json:"protocol"` //推拉流协议
Protocol int `json:"protocol"` //推拉流协议
RemoteAddr string `json:"remote_addr"` //peer地址
}
@@ -71,11 +71,11 @@ func Hook(event HookEvent, params string, body interface{}) (*http.Response, err
}
func NewHookPlayEventInfo(sink Sink) eventInfo {
return eventInfo{Stream: sink.GetSourceID(), Protocol: sink.GetProtocol().String(), RemoteAddr: sink.RemoteAddr()}
return eventInfo{Stream: sink.GetSourceID(), Protocol: int(sink.GetProtocol()), RemoteAddr: sink.RemoteAddr()}
}
func NewHookPublishEventInfo(source Source) eventInfo {
return eventInfo{Stream: source.GetID(), Protocol: source.GetType().String(), RemoteAddr: source.RemoteAddr()}
return eventInfo{Stream: source.GetID(), Protocol: int(source.GetType()), RemoteAddr: source.RemoteAddr()}
}
func NewRecordEventInfo(source Source, path string) interface{} {

View File

@@ -1,16 +1,18 @@
package stream
import (
"context"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/log"
"net/http"
"time"
)
func PreparePlaySink(sink Sink) (*http.Response, utils.HookState) {
return PreparePlaySinkWithReady(sink, true)
}
const (
ForwardSinkWaitTimeout = 20
)
func PreparePlaySinkWithReady(sink Sink, ok bool) (*http.Response, utils.HookState) {
func PreparePlaySink(sink Sink, waitTimeout bool) (*http.Response, utils.HookState) {
var response *http.Response
if AppConfig.Hooks.IsEnableOnPlay() {
@@ -24,7 +26,6 @@ func PreparePlaySinkWithReady(sink Sink, ok bool) (*http.Response, utils.HookSta
response = hook
}
sink.SetReady(ok)
source := SourceManager.Find(sink.GetSourceID())
if source == nil {
log.Sugar.Infof("添加%s sink到等待队列 id: %v source: %s", sink.GetProtocol().String(), sink.GetID(), sink.GetSourceID())
@@ -37,12 +38,22 @@ func PreparePlaySinkWithReady(sink Sink, ok bool) (*http.Response, utils.HookSta
log.Sugar.Warnf("添加到%s sink到等待队列失败, sink已经断开连接 %s", sink.GetProtocol(), sink.GetID())
return response, utils.HookStateFailure
} else {
if waitTimeout {
go func() {
timeout := sink.StartWaitTimer(context.Background(), ForwardSinkWaitTimeout*time.Second)
if timeout {
log.Sugar.Warnf("在等待队列超时, 删除%s sink id: %v source: %s", sink.GetProtocol().String(), sink.GetID(), sink.GetSourceID())
sink.Close()
}
}()
}
sink.SetState(SessionStateWaiting)
AddSinkToWaitingQueue(sink.GetSourceID(), sink)
}
}
} else {
source.AddSink(sink)
source.GetTransStreamPublisher().AddSink(sink)
}
return response, utils.HookStateOK
@@ -57,7 +68,7 @@ func HookPlayDoneEvent(sink Sink) (*http.Response, bool) {
Sink string `json:"sink"`
}{
eventInfo: NewHookPlayEventInfo(sink),
Sink: SinkId2String(sink.GetID()),
Sink: SinkID2String(sink.GetID()),
}
hook, err := Hook(HookEventPlayDone, sink.UrlValues().Encode(), body)

View File

@@ -11,19 +11,20 @@ import (
func PreparePublishSource(source Source, hook bool) (*http.Response, utils.HookState) {
var response *http.Response
if err := SourceManager.Add(source); err != nil {
return nil, utils.HookStateOccupy
}
if hook && AppConfig.Hooks.IsEnablePublishEvent() {
rep, state := HookPublishEvent(source)
if utils.HookStateOK != state {
_, _ = SourceManager.Remove(source.GetID())
return rep, state
}
response = rep
}
if err := SourceManager.Add(source); err != nil {
return nil, utils.HookStateOccupy
}
source.SetCreateTime(time.Now())
urls := GetStreamPlayUrls(source.GetID())

View File

@@ -0,0 +1,54 @@
package stream
import "github.com/lkmio/avformat/collections"
type NonBlockingChannel[T any] struct {
Channel chan T
PendingQueue *collections.LinkedList[T]
zero T
}
func (p *NonBlockingChannel[T]) Post(event T) {
for oldSize := 0; p.PendingQueue.Size() > 0 && p.PendingQueue.Size() != oldSize; {
oldSize = p.PendingQueue.Size()
select {
case p.Channel <- p.PendingQueue.Get(0):
p.PendingQueue.Remove(0)
default:
break
}
}
if p.PendingQueue.Size() != 0 {
p.PendingQueue.Add(event)
} else {
select {
case p.Channel <- event:
break
default:
p.PendingQueue.Add(event)
}
}
}
func (p *NonBlockingChannel[T]) Pop() T {
if len(p.Channel) > 0 {
select {
case event := <-p.Channel:
return event
}
}
if p.PendingQueue.Size() > 0 {
return p.PendingQueue.Remove(0)
}
return p.zero
}
func NewNonBlockingChannel[T any](size int) *NonBlockingChannel[T] {
return &NonBlockingChannel[T]{
Channel: make(chan T, size),
PendingQueue: &collections.LinkedList[T]{},
}
}

View File

@@ -54,3 +54,7 @@ func NewRtpTransStream(protocol TransStreamProtocol, capacity int) *RtpStream {
rtpBuffers: collections.NewQueue[*collections.ReferenceCounter[[]byte]](capacity),
}
}
func GBCascadedTransStreamFactory(source Source, protocol TransStreamProtocol, tracks []*Track) (TransStream, error) {
return NewRtpTransStream(TransStreamGBCascaded, 1024), nil
}

View File

@@ -98,7 +98,9 @@ type Sink interface {
// EnableAsyncWriteMode 开启异步发送
EnableAsyncWriteMode(queueSize int)
PendingSendQueueSize() int
StartWaitTimer(ctx context.Context, duration time.Duration) bool
StopWaitTimer()
}
type BaseSink struct {
@@ -124,12 +126,13 @@ type BaseSink struct {
totalDataSize atomic.Uint64
writtenDataSize atomic.Uint64
lastKeyVideoDataSegment *collections.ReferenceCounter[[]byte]
pendingSendQueue chan *collections.ReferenceCounter[[]byte] // 等待发送的数据队列
blockedBufferList *collections.LinkedList[*collections.ReferenceCounter[[]byte]] // 异步队列阻塞后的切片数据
pendingSendQueue *NonBlockingChannel[*collections.ReferenceCounter[[]byte]]
cancelFunc func()
cancelCtx context.Context
waitCtx context.Context
waitCancelFunc context.CancelFunc
}
func (s *BaseSink) GetID() SinkID {
@@ -148,8 +151,8 @@ func (s *BaseSink) fastForward(firstSegment *collections.ReferenceCounter[[]byte
firstSegment.Release()
s.writtenDataSize.Add(uint64(len(firstSegment.Get())))
for len(s.pendingSendQueue) > 0 {
buffer := <-s.pendingSendQueue
for len(s.pendingSendQueue.Channel) > 0 {
buffer := <-s.pendingSendQueue.Channel
// 不存在视频, 清空队列
// 还没有追到最近的关键帧, 继续追帧
if s.lastKeyVideoDataSegment == nil || buffer != s.lastKeyVideoDataSegment {
@@ -173,16 +176,9 @@ func (s *BaseSink) fastForward(firstSegment *collections.ReferenceCounter[[]byte
func (s *BaseSink) doAsyncWrite() {
defer func() {
// 释放未发送的数据
for len(s.pendingSendQueue) > 0 {
buffer := <-s.pendingSendQueue
for buffer := s.pendingSendQueue.Pop(); buffer != nil; buffer = s.pendingSendQueue.Pop() {
buffer.Release()
}
for s.blockedBufferList.Size() > 0 {
buffer := s.blockedBufferList.Remove(0)
buffer.Release()
}
ReleasePendingBuffers(s.SourceID, s.TransStreamID)
}()
@@ -191,7 +187,7 @@ func (s *BaseSink) doAsyncWrite() {
select {
case <-s.cancelCtx.Done():
return
case data := <-s.pendingSendQueue:
case data := <-s.pendingSendQueue.Channel:
// 追帧到最近的关键帧
if fastForward {
var ok bool
@@ -245,8 +241,7 @@ func (s *BaseSink) doAsyncWrite() {
func (s *BaseSink) EnableAsyncWriteMode(queueSize int) {
utils.Assert(s.Conn != nil)
s.pendingSendQueue = make(chan *collections.ReferenceCounter[[]byte], queueSize)
s.blockedBufferList = &collections.LinkedList[*collections.ReferenceCounter[[]byte]]{}
s.pendingSendQueue = NewNonBlockingChannel[*collections.ReferenceCounter[[]byte]](queueSize)
s.cancelCtx, s.cancelFunc = context.WithCancel(context.Background())
go s.doAsyncWrite()
}
@@ -265,50 +260,23 @@ func (s *BaseSink) Write(index int, data []*collections.ReferenceCounter[[]byte]
s.totalDataSize.Add(uint64(len(datum.Get())))
}
// 发送被阻塞的数据
for s.blockedBufferList.Size() > 0 {
bytes := s.blockedBufferList.Get(0)
select {
case s.pendingSendQueue <- bytes:
s.blockedBufferList.Remove(0)
break
default:
// 发送被阻塞的数据失败, 将本次发送的数据加入阻塞队列
for _, datum := range data {
s.blockedBufferList.Add(datum)
datum.Refer()
}
return nil
}
}
for _, bytes := range data {
if s.cancelCtx != nil {
bytes.Refer()
select {
case s.pendingSendQueue <- bytes:
break
default:
// 将本次发送的数据加入阻塞队列
s.blockedBufferList.Add(bytes)
//return transport.ZeroWindowSizeError{}
return nil
}
} else {
_, err := s.Conn.Write(bytes.Get())
if s.cancelCtx == nil {
for _, datum := range data {
_, err := s.Conn.Write(datum.Get())
if err != nil {
return err
}
}
} else {
for _, datum := range data {
datum.Refer()
s.pendingSendQueue.Post(datum)
}
}
return nil
}
func (s *BaseSink) PendingSendQueueSize() int {
return len(s.pendingSendQueue)
}
func (s *BaseSink) GetSourceID() string {
return s.SourceID
}
@@ -365,7 +333,7 @@ func (s *BaseSink) DesiredVideoCodecId() utils.AVCodecID {
// 1. Sink如果正在拉流, 删除任务交给Source处理, 否则直接从等待队列删除Sink.
// 2. 发送PlayDoneHook事件
func (s *BaseSink) Close() {
log.Sugar.Debugf("closing the %s sink. id: %s. current session state: %s", s.Protocol, SinkId2String(s.ID), s.State)
log.Sugar.Debugf("closing the %s sink. id: %s. current session state: %s", s.Protocol, SinkID2String(s.ID), s.State)
s.Lock()
defer func() {
@@ -392,7 +360,7 @@ func (s *BaseSink) Close() {
} else if s.State == SessionStateTransferring {
// 从source中删除sink, 如果source为nil, 已经结束推流.
if source := SourceManager.Find(s.SourceID); source != nil {
source.RemoveSink(s)
source.GetTransStreamPublisher().RemoveSink(s)
}
} else if s.State == SessionStateWaiting {
// 从等待队列中删除sink
@@ -472,3 +440,19 @@ func (s *BaseSink) CreateTime() time.Time {
func (s *BaseSink) SetCreateTime(time time.Time) {
s.createTime = time
}
func (s *BaseSink) StartWaitTimer(ctx context.Context, duration time.Duration) bool {
s.waitCtx, s.waitCancelFunc = context.WithCancel(ctx)
select {
case <-time.After(duration):
return true
case <-s.waitCtx.Done():
return false
}
}
func (s *BaseSink) StopWaitTimer() {
if s.waitCancelFunc != nil {
s.waitCancelFunc()
}
}

View File

@@ -1,49 +0,0 @@
package stream
import (
"fmt"
"sync"
)
// SinkManager 目前只用于保存HLS拉流Sink
var SinkManager *sinkManager
func init() {
SinkManager = &sinkManager{}
}
type sinkManager struct {
m sync.Map
}
func (s *sinkManager) Add(sink Sink) error {
_, ok := s.m.LoadOrStore(sink.GetID(), sink)
if ok {
return fmt.Errorf("the sink %s has been exist", sink.GetID())
}
return nil
}
func (s *sinkManager) Find(id SinkID) Sink {
value, ok := s.m.Load(id)
if ok {
return value.(Sink)
}
return nil
}
func (s *sinkManager) Remove(id SinkID) (Sink, error) {
value, loaded := s.m.LoadAndDelete(id)
if loaded {
return value.(Sink), nil
}
return nil, fmt.Errorf("source with GetID %s was not find", id)
}
func (s *sinkManager) Exist(id SinkID) bool {
_, ok := s.m.Load(id)
return ok
}

View File

@@ -3,8 +3,12 @@ package stream
import (
"encoding/binary"
"fmt"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/transport"
"net"
"net/url"
"strconv"
"time"
)
// SinkID IPV4使用uint64、IPV6使用string作为ID类型
@@ -18,8 +22,8 @@ func ipv4Addr2UInt64(ip uint32, port int) uint64 {
return (uint64(ip) << 32) | uint64(port)
}
// NetAddr2SinkId 根据网络地址生成SinkId IPV4使用一个uint64, IPV6使用String
func NetAddr2SinkId(addr net.Addr) SinkID {
// NetAddr2SinkID 根据网络地址生成SinkId IPV4使用一个uint64, IPV6使用String
func NetAddr2SinkID(addr net.Addr) SinkID {
network := addr.Network()
if "tcp" == network {
to4 := addr.(*net.TCPAddr).IP.To4()
@@ -42,7 +46,7 @@ func NetAddr2SinkId(addr net.Addr) SinkID {
return addr.String()
}
func SinkId2String(id SinkID) string {
func SinkID2String(id SinkID) string {
if i, ok := id.(uint64); ok {
return strconv.FormatUint(i, 10)
}
@@ -50,16 +54,53 @@ func SinkId2String(id SinkID) string {
return id.(string)
}
func GenerateUint64SinkID() SinkID {
return uint64(time.Now().UnixNano()&0xFFFFFFFF)<<32 | uint64(utils.RandomIntInRange(0, 0xFFFFFFFF))
}
func CreateSinkDisconnectionMessage(sink Sink) string {
return fmt.Sprintf("%s sink断开连接. id: %s", sink.GetProtocol(), sink.GetID())
}
func ExecuteSyncEventOnSource(sourceId string, event func()) bool {
func ExecuteSyncEventOnTransStreamPublisher(sourceId string, event func()) bool {
source := SourceManager.Find(sourceId)
if source != nil {
source.ExecuteSyncEvent(event)
source.GetTransStreamPublisher().ExecuteSyncEvent(event)
return true
}
return false
}
func SubscribeStream(sink Sink, values url.Values) utils.HookState {
return SubscribeStreamWithOptions(sink, values, true, false)
}
func SubscribeStreamWithOptions(sink Sink, values url.Values, ready bool, timeout bool) utils.HookState {
sink.SetReady(ready)
sink.SetUrlValues(values)
_, state := PreparePlaySink(sink, timeout)
return state
}
func ForwardStream(protocol TransStreamProtocol, transport TransportType, sourceId string, values url.Values, remoteAddr string, manager transport.Manager) (Sink, int, error) {
//source := SourceManager.Find(sourceId)
//if source == nil {
// return nil, 0, fmt.Errorf("source %s 不存在", sourceId)
//}
sinkId := GenerateUint64SinkID()
var port int
sink, port, err := NewForwardSink(transport, protocol, sinkId, sourceId, remoteAddr, manager)
if err != nil {
return nil, 0, err
}
state := SubscribeStreamWithOptions(sink, values, true, true)
if utils.HookStateOK != state {
sink.Close()
return nil, 0, fmt.Errorf("failed to prepare play sink")
}
return sink, port, nil
}

View File

@@ -1,6 +1,8 @@
package stream
import "sync"
import (
"sync"
)
// 等待队列所有的Sink
var waitingSinks map[string]map[SinkID]Sink
@@ -27,15 +29,21 @@ func AddSinkToWaitingQueue(streamId string, sink Sink) {
}
func RemoveSinkFromWaitingQueue(sourceId string, sinkId SinkID) (Sink, bool) {
var sink Sink
mutex.Lock()
defer mutex.Unlock()
defer func() {
mutex.Unlock()
if sink != nil {
sink.StopWaitTimer()
}
}()
m, ok := waitingSinks[sourceId]
if !ok {
return nil, false
}
sink, ok := m[sinkId]
sink, ok = m[sinkId]
if ok {
delete(m, sinkId)
}
@@ -44,15 +52,21 @@ func RemoveSinkFromWaitingQueue(sourceId string, sinkId SinkID) (Sink, bool) {
}
func PopWaitingSinks(sourceId string) []Sink {
var sinks []Sink
mutex.Lock()
defer mutex.Unlock()
defer func() {
mutex.Unlock()
for _, sink := range sinks {
sink.StopWaitTimer()
}
}()
source, ok := waitingSinks[sourceId]
if !ok {
return nil
}
sinks := make([]Sink, len(source))
sinks = make([]Sink, len(source))
var index = 0
for _, sink := range source {
sinks[index] = sink
@@ -90,13 +104,3 @@ func ExistSourceInWaitingQueue(id string) bool {
_, ok := waitingSinks[id]
return ok
}
func ExistSink(sourceId string, sinkId SinkID) bool {
if sourceId != "" {
if exist := ExistSinkInWaitingQueue(sourceId, sinkId); exist {
return true
}
}
return SinkManager.Exist(sinkId)
}

View File

@@ -5,7 +5,6 @@ import (
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/lkm/log"
"github.com/lkmio/transport"
"net"
"net/url"
"strings"
@@ -14,11 +13,10 @@ import (
"time"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/transcode"
)
var (
StreamEndInfoBride func(s Source) *StreamEndInfo
StreamEndInfoBride func(source string, tracks []*Track, streams map[TransStreamID]TransStream) *StreamEndInfo
)
// Source 对推流源的封装
@@ -39,20 +37,6 @@ type Source interface {
// OriginTracks 返回所有的推流track
OriginTracks() []*Track
// TranscodeTracks 返回所有的转码track
TranscodeTracks() []*Track
// AddSink 添加Sink, 在此之前请确保Sink已经握手、授权通过. 如果Source还未WriteHeader先将Sink添加到等待队列.
// 匹配拉流期望的编码器, 创建TransStream或向已经存在TransStream添加Sink
AddSink(sink Sink)
// RemoveSink 同步删除Sink
RemoveSink(sink Sink)
RemoveSinkWithID(id SinkID)
FindSink(id SinkID) Sink
SetState(state SessionState)
// Close 关闭Source
@@ -78,21 +62,15 @@ type Source interface {
SetUrlValues(values url.Values)
// PostEvent 切换到主协程执行当前函数
PostEvent(cb func())
postEvent(cb func())
ExecuteSyncEvent(cb func())
executeSyncEvent(cb func())
// LastPacketTime 返回最近收流时间戳
LastPacketTime() time.Time
SetLastPacketTime(time2 time.Time)
// SinkCount 返回拉流计数
SinkCount() int
// LastStreamEndTime 返回最近结束拉流时间戳
LastStreamEndTime() time.Time
IsClosed() bool
StreamPipe() chan []byte
@@ -103,15 +81,11 @@ type Source interface {
SetCreateTime(time time.Time)
Sinks() []Sink
GetBitrateStatistics() *BitrateStatistics
GetTransStreams() map[TransStreamID]TransStream
GetStreamEndInfo() *StreamEndInfo
ProbeTimeout()
GetTransStreamPublisher() TransStreamPublisher
}
type PublishSource struct {
@@ -120,38 +94,22 @@ type PublishSource struct {
state SessionState
Conn net.Conn
TransDemuxer avformat.Demuxer // 负责从推流协议中解析出AVStream和AVPacket
recordSink Sink // 每个Source的录制流
recordFilePath string // 录制流文件路径
hlsStream TransStream // HLS传输流, 如果开启, 在@see writeHeader 函数中直接创建, 如果等拉流时再创建, 会进一步加大HLS延迟.
audioTranscoders []transcode.Transcoder // 音频解码器
videoTranscoders []transcode.Transcoder // 视频解码器
originTracks TrackManager // 推流的音视频Streams
allStreamTracks TrackManager // 推流Streams+转码器获得的Stream
gopBuffer GOPBuffer // GOP缓存, 音频和视频混合使用, 以视频关键帧为界, 缓存第二个视频关键帧时, 释放前一组gop. 如果不存在视频流, 不缓存音频
streamPipe *NonBlockingChannel[[]byte] // 推流数据管道
mainContextEvents chan func() // 切换到主协程执行函数的事件管道
streamPublisher TransStreamPublisher // 解析出来的AVStream和AVPacket, 交由streamPublisher处理
closed atomic.Bool // source是否已经关闭
completed atomic.Bool // 所有推流track是否解析完毕, @see writeHeader 函数中赋值为true
TransDemuxer avformat.Demuxer // 负责从推流协议中解析出AVStream和AVPacket
originTracks TrackManager // 推流的音视频Streams
closed atomic.Bool // 是否已经关闭
completed atomic.Bool // 推流track是否解析完毕, @see writeHeader 函数中赋值为true
existVideo bool // 是否存在视频
TransStreams map[TransStreamID]TransStream // 所有输出流
ForwardTransStream TransStream // 转发流
sinks map[SinkID]Sink // 保存所有Sink
TransStreamSinks map[TransStreamID]map[SinkID]Sink // 输出流对应的Sink
streamEndInfo *StreamEndInfo // 之前推流源信息
accumulateTimestamps bool // 是否累加时间戳
timestampModeDecided bool // 是否已经决定使用推流的时间戳,或者累加时间戳
streamPipe chan []byte // 推流数据管道
mainContextEvents chan func() // 切换到主协程执行函数的事件管道
lastPacketTime time.Time // 最近收到推流包的时间
lastStreamEndTime time.Time // 最近拉流端结束拉流的时间
sinkCount int // 拉流端计数
urlValues url.Values // 推流url携带的参数
createTime time.Time // source创建时间
statistics *BitrateStatistics // 码流统计
streamLogger avformat.OnUnpackStream2FileHandler
lastPacketTime time.Time // 最近收到推流包的时间
urlValues url.Values // 推流url携带的参数
createTime time.Time // source创建时间
statistics *BitrateStatistics // 码流统计
streamLogger avformat.OnUnpackStream2FileHandler
}
func (s *PublishSource) SetLastPacketTime(time2 time.Time) {
@@ -163,31 +121,26 @@ func (s *PublishSource) IsClosed() bool {
}
func (s *PublishSource) StreamPipe() chan []byte {
return s.streamPipe
return s.streamPipe.Channel
}
func (s *PublishSource) MainContextEvents() chan func() {
return s.mainContextEvents
}
func (s *PublishSource) LastStreamEndTime() time.Time {
return s.lastStreamEndTime
}
func (s *PublishSource) LastPacketTime() time.Time {
return s.lastPacketTime
}
func (s *PublishSource) SinkCount() int {
return s.sinkCount
}
func (s *PublishSource) GetID() string {
return s.ID
}
func (s *PublishSource) SetID(id string) {
s.ID = id
if s.streamPublisher != nil {
s.streamPublisher.SetSourceID(id)
}
}
func (s *PublishSource) Init(receiveQueueSize int) {
@@ -195,52 +148,16 @@ func (s *PublishSource) Init(receiveQueueSize int) {
// 初始化事件接收管道
// -2是为了保证从管道取到流, 到处理完流整个过程安全的, 不会被覆盖
s.streamPipe = make(chan []byte, receiveQueueSize-2)
s.streamPipe = NewNonBlockingChannel[[]byte](receiveQueueSize - 1)
s.mainContextEvents = make(chan func(), 128)
s.TransStreams = make(map[TransStreamID]TransStream, 10)
s.sinks = make(map[SinkID]Sink, 128)
s.TransStreamSinks = make(map[TransStreamID]map[SinkID]Sink, len(transStreamFactories)+1)
s.statistics = NewBitrateStatistics()
s.streamPublisher = NewTransStreamPublisher(s.ID)
// 设置探测时长
s.TransDemuxer.SetProbeDuration(AppConfig.ProbeTimeout)
}
func (s *PublishSource) CreateDefaultOutStreams() {
if s.TransStreams == nil {
s.TransStreams = make(map[TransStreamID]TransStream, 10)
}
// 创建录制流
if AppConfig.Record.Enable {
sink, path, err := CreateRecordStream(s.ID)
if err != nil {
log.Sugar.Errorf("创建录制sink失败 source:%s err:%s", s.ID, err.Error())
} else {
s.recordSink = sink
s.recordFilePath = path
}
}
// 创建HLS输出流
if AppConfig.Hls.Enable {
streams := s.OriginTracks()
utils.Assert(len(streams) > 0)
id := GenerateTransStreamID(TransStreamHls, streams...)
hlsStream, err := s.CreateTransStream(id, TransStreamHls, streams)
if err != nil {
panic(err)
}
s.DispatchGOPBuffer(hlsStream)
s.hlsStream = hlsStream
s.TransStreams[id] = s.hlsStream
}
}
func (s *PublishSource) Input(data []byte) error {
s.streamPipe <- data
s.streamPipe.Post(data)
s.statistics.Input(len(data))
return nil
}
@@ -249,300 +166,6 @@ func (s *PublishSource) OriginTracks() []*Track {
return s.originTracks.All()
}
func (s *PublishSource) TranscodeTracks() []*Track {
return s.allStreamTracks.All()
}
func IsSupportMux(protocol TransStreamProtocol, _, _ utils.AVCodecID) bool {
if TransStreamRtmp == protocol || TransStreamFlv == protocol {
}
return true
}
func (s *PublishSource) CreateTransStream(id TransStreamID, protocol TransStreamProtocol, tracks []*Track) (TransStream, error) {
log.Sugar.Infof("创建%s-stream source: %s", protocol.String(), s.ID)
source := SourceManager.Find(s.ID)
utils.Assert(source != nil)
transStream, err := CreateTransStream(source, protocol, tracks)
if err != nil {
log.Sugar.Errorf("创建传输流失败 err: %s source: %s", err.Error(), s.ID)
return nil, err
}
for _, track := range tracks {
// 重新拷贝一个track传输流内部使用track的时间戳
newTrack := *track
if err = transStream.AddTrack(&newTrack); err != nil {
return nil, err
}
}
transStream.SetID(id)
transStream.SetProtocol(protocol)
// 创建输出流对应的拉流队列
s.TransStreamSinks[id] = make(map[SinkID]Sink, 128)
_ = transStream.WriteHeader()
// 设置转发流
if TransStreamGBCascadedForward == transStream.GetProtocol() {
s.ForwardTransStream = transStream
}
return transStream, err
}
func (s *PublishSource) DispatchGOPBuffer(transStream TransStream) {
if s.gopBuffer != nil {
s.gopBuffer.PeekAll(func(packet *avformat.AVPacket) {
s.DispatchPacket(transStream, packet)
})
}
}
// DispatchPacket 分发AVPacket
func (s *PublishSource) DispatchPacket(transStream TransStream, packet *avformat.AVPacket) {
data, timestamp, videoKey, err := transStream.Input(packet)
if err != nil || len(data) < 1 {
return
}
s.DispatchBuffer(transStream, packet.Index, data, timestamp, videoKey)
}
// DispatchBuffer 分发传输流
func (s *PublishSource) DispatchBuffer(transStream TransStream, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) {
sinks := s.TransStreamSinks[transStream.GetID()]
exist := transStream.IsExistVideo()
for _, sink := range sinks {
if sink.GetSentPacketCount() < 1 {
// 如果存在视频, 确保向sink发送的第一帧是关键帧
if exist && !keyVideo {
continue
}
if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
if ok := s.write(sink, index, extraData, timestamp, false); !ok {
continue
}
}
}
if ok := s.write(sink, index, data, timestamp, keyVideo); !ok {
continue
}
}
}
func (s *PublishSource) pendingSink(sink Sink) {
log.Sugar.Errorf("向sink推流超时,关闭连接. %s-sink: %s source: %s", sink.GetProtocol().String(), sink.GetID(), s.ID)
go sink.Close()
}
// 向sink推流
func (s *PublishSource) write(sink Sink, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) bool {
err := sink.Write(index, data, timestamp, keyVideo)
if err == nil {
sink.IncreaseSentPacketCount()
return true
}
// 推流超时, 可能是服务器或拉流端带宽不够、拉流端不读取数据等情况造成内核发送缓冲区满, 进而阻塞.
// 直接关闭连接. 当然也可以将sink先挂起, 后续再继续推流.
if _, ok := err.(transport.ZeroWindowSizeError); ok {
s.pendingSink(sink)
}
return false
}
// 创建sink需要的输出流
func (s *PublishSource) doAddSink(sink Sink, resume bool) bool {
// 暂时不考虑多路视频流意味着只能1路视频流和多路音频流同理originStreams和allStreams里面的Stream互斥. 同时多路音频流的Codec必须一致
audioCodecId, videoCodecId := sink.DesiredAudioCodecId(), sink.DesiredVideoCodecId()
audioTrack := s.originTracks.FindWithType(utils.AVMediaTypeAudio)
videoTrack := s.originTracks.FindWithType(utils.AVMediaTypeVideo)
disableAudio := audioTrack == nil
disableVideo := videoTrack == nil || !sink.EnableVideo()
if disableAudio && disableVideo {
return false
}
// 不支持对期望编码的流封装. 降级
if (utils.AVCodecIdNONE != audioCodecId || utils.AVCodecIdNONE != videoCodecId) && !IsSupportMux(sink.GetProtocol(), audioCodecId, videoCodecId) {
audioCodecId = utils.AVCodecIdNONE
videoCodecId = utils.AVCodecIdNONE
}
if !disableAudio && utils.AVCodecIdNONE == audioCodecId {
audioCodecId = audioTrack.Stream.CodecID
}
if !disableVideo && utils.AVCodecIdNONE == videoCodecId {
videoCodecId = videoTrack.Stream.CodecID
}
// 创建音频转码器
if !disableAudio && audioCodecId != audioTrack.Stream.CodecID {
utils.Assert(false)
}
// 创建视频转码器
if !disableVideo && videoCodecId != videoTrack.Stream.CodecID {
utils.Assert(false)
}
// 查找传输流需要的所有track
var tracks []*Track
for _, track := range s.originTracks.All() {
if disableVideo && track.Stream.MediaType == utils.AVMediaTypeVideo {
continue
}
tracks = append(tracks, track)
}
transStreamId := GenerateTransStreamID(sink.GetProtocol(), tracks...)
transStream, exist := s.TransStreams[transStreamId]
if !exist {
var err error
transStream, err = s.CreateTransStream(transStreamId, sink.GetProtocol(), tracks)
if err != nil {
log.Sugar.Errorf("添加sink失败,创建传输流发生err: %s source: %s", err.Error(), s.ID)
return false
}
s.TransStreams[transStreamId] = transStream
}
sink.SetTransStreamID(transStreamId)
{
sink.Lock()
defer sink.UnLock()
if SessionStateClosed == sink.GetState() {
log.Sugar.Warnf("添加sink失败, sink已经断开连接 %s", sink.String())
return false
} else {
sink.SetState(SessionStateTransferring)
}
}
err := sink.StartStreaming(transStream)
if err != nil {
log.Sugar.Errorf("添加sink失败,开始推流发生err: %s sink: %s source: %s ", err.Error(), SinkId2String(sink.GetID()), s.ID)
return false
}
// 还没做好准备(rtsp拉流还在协商sdp中), 暂不推流
if !sink.IsReady() {
return true
}
// 累加拉流计数
if !resume && s.recordSink != sink {
s.sinkCount++
log.Sugar.Infof("sink count: %d source: %s", s.sinkCount, s.ID)
}
s.sinks[sink.GetID()] = sink
s.TransStreamSinks[transStreamId][sink.GetID()] = sink
// TCP拉流开启异步发包, 一旦出现网络不好的链路, 其余正常链路不受影响.
_, ok := sink.GetConn().(*transport.Conn)
if ok && sink.IsTCPStreaming() {
sink.EnableAsyncWriteMode(24)
}
// 发送已有的缓存数据
// 此处发送缓存数据必须要存在关键帧的输出流才发否则等DispatchPacket时再发送extra。
data, timestamp, _ := transStream.ReadKeyFrameBuffer()
if len(data) > 0 {
if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
s.write(sink, 0, extraData, timestamp, false)
}
s.write(sink, 0, data, timestamp, true)
}
// 新建传输流,发送已经缓存的音视频帧
if !exist && AppConfig.GOPCache && s.existVideo && TransStreamGBCascadedForward != transStream.GetProtocol() {
s.DispatchGOPBuffer(transStream)
}
return true
}
func (s *PublishSource) AddSink(sink Sink) {
s.PostEvent(func() {
if !s.completed.Load() {
AddSinkToWaitingQueue(sink.GetSourceID(), sink)
} else {
if !s.doAddSink(sink, false) {
go sink.Close()
}
}
})
}
func (s *PublishSource) RemoveSink(sink Sink) {
s.ExecuteSyncEvent(func() {
s.doRemoveSink(sink)
})
}
func (s *PublishSource) RemoveSinkWithID(id SinkID) {
s.PostEvent(func() {
sink, ok := s.sinks[id]
if ok {
s.doRemoveSink(sink)
}
})
}
func (s *PublishSource) FindSink(id SinkID) Sink {
var result Sink
s.ExecuteSyncEvent(func() {
sink, ok := s.sinks[id]
if ok {
result = sink
}
})
return result
}
func (s *PublishSource) cleanupSinkStreaming(sink Sink) {
transStreamSinks := s.TransStreamSinks[sink.GetTransStreamID()]
delete(transStreamSinks, sink.GetID())
s.lastStreamEndTime = time.Now()
if sink.GetProtocol() == TransStreamHls {
// 从HLS拉流队列删除Sink
_, _ = SinkManager.Remove(sink.GetID())
}
sink.StopStreaming(s.TransStreams[sink.GetTransStreamID()])
}
func (s *PublishSource) doRemoveSink(sink Sink) bool {
s.cleanupSinkStreaming(sink)
delete(s.sinks, sink.GetID())
s.sinkCount--
log.Sugar.Infof("sink count: %d source: %s", s.sinkCount, s.ID)
utils.Assert(s.sinkCount > -1)
HookPlayDoneEvent(sink)
return true
}
func (s *PublishSource) SetState(state SessionState) {
s.state = state
}
@@ -556,24 +179,14 @@ func (s *PublishSource) DoClose() {
s.closed.Store(true)
// 释放GOP缓存
if s.gopBuffer != nil {
s.gopBuffer.PopAll(func(packet *avformat.AVPacket) {
s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
})
s.gopBuffer = nil
}
// 关闭推流源的解复用器
// 关闭推流源的解复用器, 不再接收数据
if s.TransDemuxer != nil {
s.TransDemuxer.Close()
s.TransDemuxer = nil
}
// 关闭录制流
if s.recordSink != nil {
s.recordSink.Close()
}
// 等传输流发布器关闭结束
s.streamPublisher.close()
// 释放解复用器
// 释放转码器
@@ -581,62 +194,10 @@ func (s *PublishSource) DoClose() {
_, err := SourceManager.Remove(s.ID)
if err != nil {
// source不存在, 在创建source时, 未添加到manager中, 目前只有1078流会出现这种情况(tcp连接到端口, 没有推流或推流数据无效, 无法定位到手机号, 以至于无法执行PreparePublishSource函数), 将不再处理后续事情.
log.Sugar.Errorf("删除源失败 source:%s err:%s", s.ID, err.Error())
log.Sugar.Errorf("删除源失败 source: %s err: %s", s.ID, err.Error())
return
}
// 保留推流信息
if s.sinkCount > 0 && len(s.originTracks.All()) > 0 {
sourceHistory := StreamEndInfoBride(s)
streamEndInfoManager.Add(sourceHistory)
}
// 关闭所有输出流
for _, transStream := range s.TransStreams {
// 发送剩余包
data, ts, _ := transStream.Close()
if len(data) > 0 {
s.DispatchBuffer(transStream, -1, data, ts, true)
}
// 如果是tcp传输流, 归还合并写缓冲区
if !transStream.IsTCPStreaming() || transStream.GetMWBuffer() == nil {
continue
} else if buffers := transStream.GetMWBuffer().Close(); buffers != nil {
AddMWBuffersToPending(s.ID, transStream.GetID(), buffers)
}
}
// 将所有sink添加到等待队列
for _, sink := range s.sinks {
transStreamID := sink.GetTransStreamID()
sink.SetTransStreamID(0)
if s.recordSink == sink {
continue
}
{
sink.Lock()
if SessionStateClosed == sink.GetState() {
log.Sugar.Warnf("添加到sink到等待队列失败, sink已经断开连接 %s", sink.String())
} else {
sink.SetState(SessionStateWaiting)
AddSinkToWaitingQueue(s.ID, sink)
}
sink.UnLock()
}
if SessionStateClosed != sink.GetState() {
sink.StopStreaming(s.TransStreams[transStreamID])
}
}
s.TransStreams = nil
s.sinks = nil
s.TransStreamSinks = nil
// 异步hook
go func() {
if s.Conn != nil {
@@ -645,10 +206,6 @@ func (s *PublishSource) DoClose() {
}
HookPublishDoneEvent(s)
if s.recordSink != nil {
HookRecordEvent(s, s.recordFilePath)
}
}()
}
@@ -658,7 +215,7 @@ func (s *PublishSource) Close() {
}
// 同步执行, 确保close后, 主协程已经退出, 不会再处理任何推拉流、查询等任何事情.
s.ExecuteSyncEvent(func() {
s.executeSyncEvent(func() {
s.DoClose()
})
}
@@ -672,46 +229,13 @@ func (s *PublishSource) writeHeader() {
s.completed.Store(true)
s.streamPublisher.Post(&StreamEvent{StreamEventTypeTrackCompleted, nil})
if len(s.originTracks.All()) == 0 {
log.Sugar.Errorf("没有一路track, 删除source: %s", s.ID)
s.DoClose()
return
}
// 尝试恢复上次推流的会话
if streamInfo := streamEndInfoManager.Remove(s.ID); streamInfo != nil && EqualsTracks(streamInfo, s.originTracks.All()) {
s.streamEndInfo = streamInfo
// 恢复每路track的时间戳
tracks := s.originTracks.All()
for _, track := range tracks {
timestamps := streamInfo.Timestamps[track.Stream.CodecID]
track.Dts = timestamps[0]
track.Pts = timestamps[1]
}
}
// 纠正GOP中的时间戳
if s.gopBuffer != nil && s.gopBuffer.Size() != 0 {
s.gopBuffer.PeekAll(func(packet *avformat.AVPacket) {
s.CorrectTimestamp(packet)
})
}
// 创建录制流和HLS
s.CreateDefaultOutStreams()
// 将等待队列的sink添加到输出流队列
sinks := PopWaitingSinks(s.ID)
if s.recordSink != nil {
sinks = append(sinks, s.recordSink)
}
for _, sink := range sinks {
if !s.doAddSink(sink, false) {
go sink.Close()
}
}
}
func (s *PublishSource) IsCompleted() bool {
@@ -729,31 +253,6 @@ func (s *PublishSource) NotTrackAdded(index int) bool {
return true
}
func (s *PublishSource) CorrectTimestamp(packet *avformat.AVPacket) {
// 对比第一包的时间戳和上次推流的最后时间戳。如果小于上次的推流时间戳,则在原来的基础上累加。
if s.streamEndInfo != nil && !s.timestampModeDecided {
s.timestampModeDecided = true
timestamps := s.streamEndInfo.Timestamps[packet.CodecID]
s.accumulateTimestamps = true
log.Sugar.Infof("累加时间戳 上次推流dts: %d, pts: %d", timestamps[0], timestamps[1])
}
track := s.originTracks.Find(packet.CodecID)
duration := packet.GetDuration(packet.Timebase)
// 根据duration来累加时间戳
if s.accumulateTimestamps {
offset := packet.Pts - packet.Dts
packet.Dts = track.Dts + duration
packet.Pts = packet.Dts + offset
}
track.Dts = packet.Dts
track.Pts = packet.Pts
track.FrameDuration = int(duration)
}
func (s *PublishSource) OnNewTrack(track avformat.Track) {
if AppConfig.Debug {
s.streamLogger.Path = "dump/" + strings.ReplaceAll(s.ID, "/", "_")
@@ -770,17 +269,14 @@ func (s *PublishSource) OnNewTrack(track avformat.Track) {
return
}
s.originTracks.Add(NewTrack(stream, 0, 0))
s.allStreamTracks.Add(NewTrack(stream, 0, 0))
newTrack := NewTrack(stream, 0, 0)
s.originTracks.Add(newTrack)
if utils.AVMediaTypeVideo == stream.MediaType {
s.existVideo = true
}
// 创建GOPBuffer
if AppConfig.GOPCache && s.existVideo && s.gopBuffer == nil {
s.gopBuffer = NewStreamBuffer()
}
s.streamPublisher.Post(&StreamEvent{StreamEventTypeTrack, newTrack})
}
func (s *PublishSource) OnTrackComplete() {
@@ -810,32 +306,20 @@ func (s *PublishSource) OnPacket(packet *avformat.AVPacket) {
return
}
// 保存到GOP缓存
if AppConfig.GOPCache && s.existVideo {
// GOP队列溢出
if s.gopBuffer.RequiresClear(packet) {
s.gopBuffer.PopAll(func(packet *avformat.AVPacket) {
s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
})
packetPtr := collections.NewReferenceCounter(packet)
packetPtr.Refer() // 引用计数加1
packets := s.originTracks.FindWithType(packet.MediaType).Packets
packets.Add(packetPtr)
s.streamPublisher.Post(&StreamEvent{StreamEventTypePacket, packetPtr})
// 释放未引用的AVPacket
for packets.Size() > 0 {
if packets.Get(0).UseCount() > 1 {
break
}
s.gopBuffer.AddPacket(packet)
}
// track解析完毕后才能生成传输流
if s.completed.Load() {
s.CorrectTimestamp(packet)
// 分发给各个传输流
for _, transStream := range s.TransStreams {
if TransStreamGBCascadedForward != transStream.GetProtocol() {
s.DispatchPacket(transStream, packet)
}
}
}
// 未开启GOP缓存或只存在音频流, 立即释放
if !AppConfig.GOPCache || !s.existVideo {
packets.Remove(0).Release()
s.TransDemuxer.DiscardHeadPacket(packet.BufferIndex)
}
}
@@ -872,15 +356,15 @@ func (s *PublishSource) SetUrlValues(values url.Values) {
s.urlValues = values
}
func (s *PublishSource) PostEvent(cb func()) {
func (s *PublishSource) postEvent(cb func()) {
s.mainContextEvents <- cb
}
func (s *PublishSource) ExecuteSyncEvent(cb func()) {
func (s *PublishSource) executeSyncEvent(cb func()) {
group := sync.WaitGroup{}
group.Add(1)
s.PostEvent(func() {
s.postEvent(func() {
cb()
group.Done()
})
@@ -896,32 +380,16 @@ func (s *PublishSource) SetCreateTime(time time.Time) {
s.createTime = time
}
func (s *PublishSource) Sinks() []Sink {
var sinks []Sink
s.ExecuteSyncEvent(func() {
for _, sink := range s.sinks {
sinks = append(sinks, sink)
}
})
return sinks
}
func (s *PublishSource) GetBitrateStatistics() *BitrateStatistics {
return s.statistics
}
func (s *PublishSource) GetTransStreams() map[TransStreamID]TransStream {
return s.TransStreams
}
func (s *PublishSource) GetStreamEndInfo() *StreamEndInfo {
return s.streamEndInfo
}
func (s *PublishSource) ProbeTimeout() {
if s.TransDemuxer != nil {
s.TransDemuxer.ProbeComplete()
}
}
func (s *PublishSource) GetTransStreamPublisher() TransStreamPublisher {
return s.streamPublisher
}

View File

@@ -25,13 +25,14 @@ const (
SourceType1078 = SourceType(3)
SourceTypeGBTalk = SourceType(4) // 国标广播/对讲
TransStreamRtmp = TransStreamProtocol(1)
TransStreamFlv = TransStreamProtocol(2)
TransStreamRtsp = TransStreamProtocol(3)
TransStreamHls = TransStreamProtocol(4)
TransStreamRtc = TransStreamProtocol(5)
TransStreamGBCascadedForward = TransStreamProtocol(6) // 国标级联转发
TransStreamGBTalkForward = TransStreamProtocol(7) // 国标广播/对讲转发
TransStreamRtmp = TransStreamProtocol(1)
TransStreamFlv = TransStreamProtocol(2)
TransStreamRtsp = TransStreamProtocol(3)
TransStreamHls = TransStreamProtocol(4)
TransStreamRtc = TransStreamProtocol(5)
TransStreamGBCascaded = TransStreamProtocol(6) // 国标级联转发
TransStreamGBTalk = TransStreamProtocol(7) // 国标广播/对讲转发
TransStreamGBGateway = TransStreamProtocol(8) // 国标网关
)
const (
@@ -48,7 +49,7 @@ func (s SourceType) String() string {
if SourceTypeRtmp == s {
return "rtmp"
} else if SourceType28181 == s {
return "28181"
return "gb28181"
} else if SourceType1078 == s {
return "jt1078"
} else if SourceTypeGBTalk == s {
@@ -69,10 +70,12 @@ func (p TransStreamProtocol) String() string {
return "hls"
} else if TransStreamRtc == p {
return "rtc"
} else if TransStreamGBCascadedForward == p {
return "gb_cascaded_forward"
} else if TransStreamGBTalkForward == p {
return "gb_talk_forward"
} else if TransStreamGBCascaded == p {
return "gb_cascaded"
} else if TransStreamGBTalk == p {
return "gb_talk"
} else if TransStreamGBGateway == p {
return "gb_gateway"
}
panic(fmt.Sprintf("unknown stream protocol %d", p))
@@ -133,97 +136,6 @@ func ParseUrl(name string) (string, url.Values) {
return name, nil
}
//
//func ExtractVideoPacket(codec utils.AVCodecID, key, extractStream bool, data []byte, pts, dts int64, index, timebase int) (*avformat.AVStream, *avformat.AVPacket, error) {
// var stream *avformat.AVStream
//
// if utils.AVCodecIdH264 == codec {
// //从关键帧中解析出sps和pps
// if key && extractStream {
// sps, pps, err := avc.ParseExtraDataFromKeyNALU(data)
// if err != nil {
// log.Sugar.Errorf("从关键帧中解析sps pps失败 data:%s", hex.EncodeToString(data))
// return nil, nil, err
// }
//
// codecData, err := utils.NewAVCCodecData(sps, pps)
// if err != nil {
// log.Sugar.Errorf("解析sps pps失败 data:%s sps:%s, pps:%s", hex.EncodeToString(data), hex.EncodeToString(sps), hex.EncodeToString(pps))
// return nil, nil, err
// }
//
// stream = avformat.NewAVStream(utils.AVMediaTypeVideo, 0, codec, codecData.AnnexBExtraData(), codecData)
// }
//
// } else if utils.AVCodecIdH265 == codec {
// if key && extractStream {
// vps, sps, pps, err := hevc.ParseExtraDataFromKeyNALU(data)
// if err != nil {
// log.Sugar.Errorf("从关键帧中解析vps sps pps失败 data:%s", hex.EncodeToString(data))
// return nil, nil, err
// }
//
// codecData, err := utils.NewHEVCCodecData(vps, sps, pps)
// if err != nil {
// log.Sugar.Errorf("解析sps pps失败 data:%s vps:%s sps:%s, pps:%s", hex.EncodeToString(data), hex.EncodeToString(vps), hex.EncodeToString(sps), hex.EncodeToString(pps))
// return nil, nil, err
// }
//
// stream = avformat.NewAVStream(utils.AVMediaTypeVideo, 0, codec, codecData.AnnexBExtraData(), codecData)
// }
//
// }
//
// packet := avformat.NewVideoPacket(data, dts, pts, key, utils.PacketTypeAnnexB, codec, index, timebase)
// return stream, packet, nil
//}
//
//func ExtractAudioPacket(codec utils.AVCodecID, extractStream bool, data []byte, pts, dts int64, index, timebase int) (*avformat.AVStream, *avformat.AVPacket, error) {
// var stream *avformat.AVStream
// var packet *avformat.AVPacket
// if utils.AVCodecIdAAC == codec {
// //必须包含ADTSHeader
// if len(data) < 7 {
// return nil, nil, fmt.Errorf("need more data")
// }
//
// var skip int
// header, err := utils.ReadADtsFixedHeader(data)
// if err != nil {
// log.Sugar.Errorf("读取ADTSHeader失败 data:%s", hex.EncodeToString(data[:7]))
// return nil, nil, err
// } else {
// skip = 7
// //跳过ADtsHeader长度
// if header.ProtectionAbsent() == 0 {
// skip += 2
// }
// }
//
// if extractStream {
// configData, err := utils.ADtsHeader2MpegAudioConfigData(header)
// config, err := utils.ParseMpeg4AudioConfig(configData)
// println(config)
// if err != nil {
// log.Sugar.Errorf("adt头转m4ac失败 data:%s", hex.EncodeToString(data[:7]))
// return nil, nil, err
// }
//
// stream = avformat.NewAVStream(utils.AVMediaTypeAudio, index, codec, configData, nil)
// }
//
// packet = utils.NewAudioPacket(data[skip:], dts, pts, codec, index, timebase)
// } else if utils.AVCodecIdPCMALAW == codec || utils.AVCodecIdPCMMULAW == codec {
// if extractStream {
// stream = avformat.NewAVStream(utils.AVMediaTypeAudio, index, codec, nil, nil)
// }
//
// packet = utils.NewAudioPacket(data, dts, pts, codec, index, timebase)
// }
//
// return stream, packet, nil
//}
// StartReceiveDataTimer 启动收流超时计时器
// 收流超时, 客观上认为是流中断, 应该关闭Source. 如果开启了Hook, 并且Hook返回200应答, 则不关闭Source.
func StartReceiveDataTimer(source Source) *time.Timer {
@@ -264,9 +176,9 @@ func StartIdleTimer(source Source) *time.Timer {
var idleTimer *time.Timer
idleTimer = time.AfterFunc(time.Duration(AppConfig.IdleTimeout), func() {
dis := time.Now().Sub(source.LastStreamEndTime())
dis := time.Now().Sub(source.GetTransStreamPublisher().LastStreamEndTime())
if source.SinkCount() < 1 && dis >= time.Duration(AppConfig.IdleTimeout) {
if source.GetTransStreamPublisher().SinkCount() < 1 && dis >= time.Duration(AppConfig.IdleTimeout) {
log.Sugar.Errorf("拉流空闲超时 source: %s", source.GetID())
// 此处不参考返回值err, 客观希望不关闭Source
@@ -334,7 +246,7 @@ func LoopEvent(source Source) {
}
var ok bool
source.ExecuteSyncEvent(func() {
source.executeSyncEvent(func() {
source.ProbeTimeout()
ok = len(source.OriginTracks()) > 0
})
@@ -345,6 +257,9 @@ func LoopEvent(source Source) {
}
})
// 启动协程, 生成发布传输流
go source.GetTransStreamPublisher().run()
for {
select {
// 读取推流数据

725
stream/stream_publisher.go Normal file
View File

@@ -0,0 +1,725 @@
package stream
import (
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/collections"
"github.com/lkmio/avformat/utils"
"github.com/lkmio/lkm/log"
"github.com/lkmio/lkm/transcode"
"github.com/lkmio/transport"
"sync"
"sync/atomic"
"time"
)
type StreamEventType int
const (
StreamEventTypeTrack StreamEventType = iota + 1
StreamEventTypeTrackCompleted
StreamEventTypePacket
StreamEventTypeRawPacket
)
type StreamEvent struct {
Type StreamEventType
Data interface{}
}
type TransStreamPublisher interface {
Post(event *StreamEvent)
run()
close()
Sinks() []Sink
GetTransStreams() map[TransStreamID]TransStream
GetForwardTransStream() TransStream
GetStreamEndInfo() *StreamEndInfo
// SinkCount 返回拉流计数
SinkCount() int
// LastStreamEndTime 返回最近结束拉流时间戳
LastStreamEndTime() time.Time
// TranscodeTracks 返回所有的转码track
TranscodeTracks() []*Track
// AddSink 添加Sink, 在此之前请确保Sink已经握手、授权通过. 如果Source还未WriteHeader先将Sink添加到等待队列.
// 匹配拉流期望的编码器, 创建TransStream或向已经存在TransStream添加Sink
AddSink(sink Sink)
// RemoveSink 同步删除Sink
RemoveSink(sink Sink)
RemoveSinkWithID(id SinkID)
FindSink(id SinkID) Sink
ExecuteSyncEvent(cb func())
SetSourceID(id string)
}
type transStreamPublisher struct {
source string
streamEvents *NonBlockingChannel[*StreamEvent]
mainContextEvents chan func()
sinkCount int
gopBuffer GOPBuffer // GOP缓存, 音频和视频混合使用, 以视频关键帧为界, 缓存第二个视频关键帧时, 释放前一组gop
recordSink Sink // 每个Source的录制流
recordFilePath string // 录制流文件路径
hlsStream TransStream // HLS传输流, 如果开启, 在@see writeHeader 函数中直接创建, 如果等拉流时再创建, 会进一步加大HLS延迟.
_ []transcode.Transcoder // 音频解码器
_ []transcode.Transcoder // 视频解码器
originTracks TrackManager // 推流的音视频Streams
allStreamTracks TrackManager // 推流Streams+转码器获得的Stream
transStreams map[TransStreamID]TransStream // 所有输出流
forwardTransStream TransStream // 转发流
sinks map[SinkID]Sink // 保存所有Sink
transStreamSinks map[TransStreamID]map[SinkID]Sink // 输出流对应的Sink
existVideo bool // 是否存在视频
completed atomic.Bool // 所有推流track是否解析完毕, @see writeHeader 函数中赋值为true
closed atomic.Bool
streamEndInfo *StreamEndInfo // 之前推流源信息
accumulateTimestamps bool // 是否累加时间戳
timestampModeDecided bool // 是否已经决定使用推流的时间戳,或者累加时间戳
lastStreamEndTime time.Time // 最近拉流端结束拉流的时间
}
func (t *transStreamPublisher) Post(event *StreamEvent) {
t.streamEvents.Post(event)
}
func (t *transStreamPublisher) run() {
t.streamEvents = NewNonBlockingChannel[*StreamEvent](256)
t.mainContextEvents = make(chan func(), 256)
t.transStreams = make(map[TransStreamID]TransStream, 10)
t.sinks = make(map[SinkID]Sink, 128)
t.transStreamSinks = make(map[TransStreamID]map[SinkID]Sink, len(transStreamFactories)+1)
defer func() {
// 清空管道
for event := t.streamEvents.Pop(); event != nil; event = t.streamEvents.Pop() {
if StreamEventTypePacket == event.Type {
event.Data.(*collections.ReferenceCounter[*avformat.AVPacket]).Release()
}
}
}()
for {
select {
case event := <-t.streamEvents.Channel:
switch event.Type {
case StreamEventTypeTrack:
// 添加track
t.OnNewTrack(event.Data.(*Track))
case StreamEventTypeTrackCompleted:
t.WriteHeader()
// track完成
case StreamEventTypePacket:
// 发送数据包
t.OnPacket(event.Data.(*collections.ReferenceCounter[*avformat.AVPacket]))
case StreamEventTypeRawPacket:
// 发送原始数据包, 目前仅用于国标级联转发
if t.forwardTransStream != nil && t.forwardTransStream.GetProtocol() == TransStreamGBCascaded {
packets := event.Data.([][]byte)
for _, data := range packets {
t.DispatchPacket(t.forwardTransStream, &avformat.AVPacket{Data: data[2:]})
UDPReceiveBufferPool.Put(data[:cap(data)])
}
}
}
case event := <-t.mainContextEvents:
event()
if t.closed.Load() {
return
}
}
}
}
func (t *transStreamPublisher) PostEvent(cb func()) {
t.mainContextEvents <- cb
}
func (t *transStreamPublisher) ExecuteSyncEvent(cb func()) {
group := sync.WaitGroup{}
group.Add(1)
t.PostEvent(func() {
cb()
group.Done()
})
group.Wait()
}
func (t *transStreamPublisher) CreateDefaultOutStreams() {
if t.transStreams == nil {
t.transStreams = make(map[TransStreamID]TransStream, 10)
}
// 创建录制流
if AppConfig.Record.Enable {
sink, path, err := CreateRecordStream(t.source)
if err != nil {
log.Sugar.Errorf("创建录制sink失败 source: %s err: %s", t.source, err.Error())
} else {
t.recordSink = sink
t.recordFilePath = path
}
}
// 创建HLS输出流
if AppConfig.Hls.Enable {
streams := t.originTracks.All()
utils.Assert(len(streams) > 0)
id := GenerateTransStreamID(TransStreamHls, streams...)
hlsStream, err := t.CreateTransStream(id, TransStreamHls, streams)
if err != nil {
panic(err)
}
t.DispatchGOPBuffer(hlsStream)
t.hlsStream = hlsStream
t.transStreams[id] = t.hlsStream
}
}
func IsSupportMux(protocol TransStreamProtocol, _, _ utils.AVCodecID) bool {
if TransStreamRtmp == protocol || TransStreamFlv == protocol {
}
return true
}
func (t *transStreamPublisher) CreateTransStream(id TransStreamID, protocol TransStreamProtocol, tracks []*Track) (TransStream, error) {
log.Sugar.Infof("创建%s-stream source: %s", protocol.String(), t.source)
source := SourceManager.Find(t.source)
utils.Assert(source != nil)
transStream, err := CreateTransStream(source, protocol, tracks)
if err != nil {
log.Sugar.Errorf("创建传输流失败 err: %s source: %s", err.Error(), t.source)
return nil, err
}
for _, track := range tracks {
// 重新拷贝一个track传输流内部使用track的时间戳
newTrack := *track
if err = transStream.AddTrack(&newTrack); err != nil {
return nil, err
}
}
transStream.SetID(id)
transStream.SetProtocol(protocol)
// 创建输出流对应的拉流队列
t.transStreamSinks[id] = make(map[SinkID]Sink, 128)
_ = transStream.WriteHeader()
// 设置转发流
if TransStreamGBCascaded == transStream.GetProtocol() {
t.forwardTransStream = transStream
}
return transStream, err
}
func (t *transStreamPublisher) DispatchGOPBuffer(transStream TransStream) {
if t.gopBuffer != nil {
t.gopBuffer.PeekAll(func(packet *collections.ReferenceCounter[*avformat.AVPacket]) {
t.DispatchPacket(transStream, packet.Get())
})
}
}
// DispatchPacket 分发AVPacket
func (t *transStreamPublisher) DispatchPacket(transStream TransStream, packet *avformat.AVPacket) {
data, timestamp, videoKey, err := transStream.Input(packet)
if err != nil || len(data) < 1 {
return
}
t.DispatchBuffer(transStream, packet.Index, data, timestamp, videoKey)
}
// DispatchBuffer 分发传输流
func (t *transStreamPublisher) DispatchBuffer(transStream TransStream, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) {
sinks := t.transStreamSinks[transStream.GetID()]
exist := transStream.IsExistVideo()
for _, sink := range sinks {
if sink.GetSentPacketCount() < 1 {
// 如果存在视频, 确保向sink发送的第一帧是关键帧
if exist && !keyVideo {
continue
}
if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
if ok := t.write(sink, index, extraData, timestamp, false); !ok {
continue
}
}
}
if ok := t.write(sink, index, data, timestamp, keyVideo); !ok {
continue
}
}
}
func (t *transStreamPublisher) pendingSink(sink Sink) {
log.Sugar.Errorf("向sink推流超时,关闭连接. %s-sink: %s source: %s", sink.GetProtocol().String(), sink.GetID(), t.source)
go sink.Close()
}
// 向sink推流
func (t *transStreamPublisher) write(sink Sink, index int, data []*collections.ReferenceCounter[[]byte], timestamp int64, keyVideo bool) bool {
err := sink.Write(index, data, timestamp, keyVideo)
if err == nil {
sink.IncreaseSentPacketCount()
return true
}
// 推流超时, 可能是服务器或拉流端带宽不够、拉流端不读取数据等情况造成内核发送缓冲区满, 进而阻塞.
// 直接关闭连接. 当然也可以将sink先挂起, 后续再继续推流.
if _, ok := err.(transport.ZeroWindowSizeError); ok {
t.pendingSink(sink)
}
return false
}
// 创建sink需要的输出流
func (t *transStreamPublisher) doAddSink(sink Sink, resume bool) bool {
// 暂时不考虑多路视频流意味着只能1路视频流和多路音频流同理originStreams和allStreams里面的Stream互斥. 同时多路音频流的Codec必须一致
audioCodecId, videoCodecId := sink.DesiredAudioCodecId(), sink.DesiredVideoCodecId()
audioTrack := t.originTracks.FindWithType(utils.AVMediaTypeAudio)
videoTrack := t.originTracks.FindWithType(utils.AVMediaTypeVideo)
disableAudio := audioTrack == nil
disableVideo := videoTrack == nil || !sink.EnableVideo()
if disableAudio && disableVideo {
return false
}
// 不支持对期望编码的流封装. 降级
if (utils.AVCodecIdNONE != audioCodecId || utils.AVCodecIdNONE != videoCodecId) && !IsSupportMux(sink.GetProtocol(), audioCodecId, videoCodecId) {
audioCodecId = utils.AVCodecIdNONE
videoCodecId = utils.AVCodecIdNONE
}
if !disableAudio && utils.AVCodecIdNONE == audioCodecId {
audioCodecId = audioTrack.Stream.CodecID
}
if !disableVideo && utils.AVCodecIdNONE == videoCodecId {
videoCodecId = videoTrack.Stream.CodecID
}
// 创建音频转码器
if !disableAudio && audioCodecId != audioTrack.Stream.CodecID {
utils.Assert(false)
}
// 创建视频转码器
if !disableVideo && videoCodecId != videoTrack.Stream.CodecID {
utils.Assert(false)
}
// 查找传输流需要的所有track
var tracks []*Track
for _, track := range t.originTracks.All() {
if disableVideo && track.Stream.MediaType == utils.AVMediaTypeVideo {
continue
}
tracks = append(tracks, track)
}
transStreamId := GenerateTransStreamID(sink.GetProtocol(), tracks...)
transStream, exist := t.transStreams[transStreamId]
if !exist {
var err error
transStream, err = t.CreateTransStream(transStreamId, sink.GetProtocol(), tracks)
if err != nil {
log.Sugar.Errorf("添加sink失败,创建传输流发生err: %s source: %s", err.Error(), t.source)
return false
}
t.transStreams[transStreamId] = transStream
}
sink.SetTransStreamID(transStreamId)
{
sink.Lock()
defer sink.UnLock()
if SessionStateClosed == sink.GetState() {
log.Sugar.Warnf("添加sink失败, sink已经断开连接 %s", sink.String())
return false
} else {
sink.SetState(SessionStateTransferring)
}
}
err := sink.StartStreaming(transStream)
if err != nil {
log.Sugar.Errorf("添加sink失败,开始推流发生err: %s sink: %s source: %s ", err.Error(), SinkID2String(sink.GetID()), t.source)
return false
}
// 还没做好准备(rtsp拉流还在协商sdp中), 暂不推流
if !sink.IsReady() {
return true
}
// 累加拉流计数
if !resume && t.recordSink != sink {
t.sinkCount++
log.Sugar.Infof("sink count: %d source: %s", t.sinkCount, t.source)
}
t.sinks[sink.GetID()] = sink
t.transStreamSinks[transStreamId][sink.GetID()] = sink
// TCP拉流开启异步发包, 一旦出现网络不好的链路, 其余正常链路不受影响.
_, ok := sink.GetConn().(*transport.Conn)
if ok && sink.IsTCPStreaming() {
sink.EnableAsyncWriteMode(24)
}
// 发送已有的缓存数据
// 此处发送缓存数据必须要存在关键帧的输出流才发否则等DispatchPacket时再发送extra。
data, timestamp, _ := transStream.ReadKeyFrameBuffer()
if len(data) > 0 {
if extraData, _, _ := transStream.ReadExtraData(timestamp); len(extraData) > 0 {
t.write(sink, 0, extraData, timestamp, false)
}
t.write(sink, 0, data, timestamp, true)
}
// 新建传输流,发送已经缓存的音视频帧
if !exist && AppConfig.GOPCache && t.existVideo && TransStreamGBCascaded != transStream.GetProtocol() {
t.DispatchGOPBuffer(transStream)
}
return true
}
func (t *transStreamPublisher) AddSink(sink Sink) {
t.PostEvent(func() {
if !t.completed.Load() {
AddSinkToWaitingQueue(sink.GetSourceID(), sink)
} else {
if !t.doAddSink(sink, false) {
go sink.Close()
}
}
})
}
func (t *transStreamPublisher) RemoveSink(sink Sink) {
t.ExecuteSyncEvent(func() {
t.doRemoveSink(sink)
})
}
func (t *transStreamPublisher) RemoveSinkWithID(id SinkID) {
t.PostEvent(func() {
sink, ok := t.sinks[id]
if ok {
t.doRemoveSink(sink)
}
})
}
func (t *transStreamPublisher) FindSink(id SinkID) Sink {
var result Sink
t.ExecuteSyncEvent(func() {
sink, ok := t.sinks[id]
if ok {
result = sink
}
})
return result
}
func (t *transStreamPublisher) cleanupSinkStreaming(sink Sink) {
transStreamSinks := t.transStreamSinks[sink.GetTransStreamID()]
delete(transStreamSinks, sink.GetID())
t.lastStreamEndTime = time.Now()
sink.StopStreaming(t.transStreams[sink.GetTransStreamID()])
}
func (t *transStreamPublisher) doRemoveSink(sink Sink) bool {
t.cleanupSinkStreaming(sink)
delete(t.sinks, sink.GetID())
t.sinkCount--
log.Sugar.Infof("sink count: %d source: %s", t.sinkCount, t.source)
utils.Assert(t.sinkCount > -1)
HookPlayDoneEvent(sink)
return true
}
func (t *transStreamPublisher) close() {
t.ExecuteSyncEvent(func() {
t.doClose()
})
}
func (t *transStreamPublisher) doClose() {
t.closed.Store(true)
// 释放GOP缓存
if t.gopBuffer != nil {
t.gopBuffer.PopAll(func(packet *collections.ReferenceCounter[*avformat.AVPacket]) {
packet.Release()
})
t.gopBuffer = nil
}
// 关闭录制流
if t.recordSink != nil {
t.recordSink.Close()
}
// 保留推流信息
if t.sinkCount > 0 && len(t.originTracks.All()) > 0 {
sourceHistory := StreamEndInfoBride(t.source, t.originTracks.All(), t.transStreams)
streamEndInfoManager.Add(sourceHistory)
}
// 关闭所有输出流
for _, transStream := range t.transStreams {
// 发送剩余包
data, ts, _ := transStream.Close()
if len(data) > 0 {
t.DispatchBuffer(transStream, -1, data, ts, true)
}
// 如果是tcp传输流, 归还合并写缓冲区
if !transStream.IsTCPStreaming() || transStream.GetMWBuffer() == nil {
continue
} else if buffers := transStream.GetMWBuffer().Close(); buffers != nil {
AddMWBuffersToPending(t.source, transStream.GetID(), buffers)
}
}
// 将所有sink添加到等待队列
for _, sink := range t.sinks {
transStreamID := sink.GetTransStreamID()
sink.SetTransStreamID(0)
if t.recordSink == sink {
continue
}
{
sink.Lock()
if SessionStateClosed == sink.GetState() {
log.Sugar.Warnf("添加到sink到等待队列失败, sink已经断开连接 %s", sink.String())
} else {
sink.SetState(SessionStateWaiting)
AddSinkToWaitingQueue(t.source, sink)
}
sink.UnLock()
}
if SessionStateClosed != sink.GetState() {
sink.StopStreaming(t.transStreams[transStreamID])
}
}
t.transStreams = nil
t.sinks = nil
t.transStreamSinks = nil
}
func (t *transStreamPublisher) WriteHeader() {
t.completed.Store(true)
// 尝试使用上次结束推流的时间戳
if streamInfo := streamEndInfoManager.Remove(t.source); streamInfo != nil && EqualsTracks(streamInfo, t.originTracks.All()) {
t.streamEndInfo = streamInfo
// 恢复每路track的时间戳
tracks := t.originTracks.All()
for _, track := range tracks {
timestamps := streamInfo.Timestamps[track.Stream.CodecID]
track.Dts = timestamps[0]
track.Pts = timestamps[1]
}
}
// 纠正GOP中的时间戳
if t.gopBuffer != nil && t.gopBuffer.Size() != 0 {
t.gopBuffer.PeekAll(func(packet *collections.ReferenceCounter[*avformat.AVPacket]) {
t.CorrectTimestamp(packet.Get())
})
}
// 创建录制流和HLS
t.CreateDefaultOutStreams()
// 将等待队列的sink添加到输出流队列
sinks := PopWaitingSinks(t.source)
if t.recordSink != nil {
sinks = append(sinks, t.recordSink)
}
for _, sink := range sinks {
if !t.doAddSink(sink, false) {
go sink.Close()
}
}
// 如果不存在视频帧, 清空GOP缓存
if !t.existVideo {
t.gopBuffer.PopAll(func(c *collections.ReferenceCounter[*avformat.AVPacket]) {
c.Refer()
})
t.gopBuffer = nil
}
}
func (t *transStreamPublisher) Sinks() []Sink {
var sinks []Sink
t.ExecuteSyncEvent(func() {
for _, sink := range t.sinks {
sinks = append(sinks, sink)
}
})
return sinks
}
func (t *transStreamPublisher) OnPacket(packet *collections.ReferenceCounter[*avformat.AVPacket]) {
// 保存到GOP缓存
if (AppConfig.GOPCache && t.existVideo) || !t.completed.Load() {
// GOP队列溢出
if t.gopBuffer.RequiresClear(packet) {
t.gopBuffer.PopAll(func(packet *collections.ReferenceCounter[*avformat.AVPacket]) {
packet.Release()
})
}
t.gopBuffer.AddPacket(packet)
}
// track解析完毕后才能生成传输流
if t.completed.Load() {
t.CorrectTimestamp(packet.Get())
// 分发给各个传输流
for _, transStream := range t.transStreams {
if TransStreamGBCascaded != transStream.GetProtocol() {
t.DispatchPacket(transStream, packet.Get())
}
}
}
// 未开启GOP缓存或只存在音频流, 立即释放
if !AppConfig.GOPCache || !t.existVideo {
packet.Release()
}
}
func (t *transStreamPublisher) OnNewTrack(track *Track) {
stream := track.Stream
t.originTracks.Add(track)
if utils.AVMediaTypeVideo == stream.MediaType {
t.existVideo = true
}
// 创建GOPBuffer
if t.gopBuffer == nil {
t.gopBuffer = NewStreamBuffer()
}
}
// CorrectTimestamp 纠正时间戳
func (t *transStreamPublisher) CorrectTimestamp(packet *avformat.AVPacket) {
// 对比第一包的时间戳和上次推流的最后时间戳。如果小于上次的推流时间戳,则在原来的基础上累加。
if t.streamEndInfo != nil && !t.timestampModeDecided {
t.timestampModeDecided = true
timestamps := t.streamEndInfo.Timestamps[packet.CodecID]
t.accumulateTimestamps = true
log.Sugar.Infof("累加时间戳 上次推流dts: %d, pts: %d", timestamps[0], timestamps[1])
}
track := t.originTracks.Find(packet.CodecID)
duration := packet.GetDuration(packet.Timebase)
// 根据duration来累加时间戳
if t.accumulateTimestamps {
offset := packet.Pts - packet.Dts
packet.Dts = track.Dts + duration
packet.Pts = packet.Dts + offset
}
track.Dts = packet.Dts
track.Pts = packet.Pts
track.FrameDuration = int(duration)
}
func (t *transStreamPublisher) GetTransStreams() map[TransStreamID]TransStream {
return t.transStreams
}
func (t *transStreamPublisher) GetStreamEndInfo() *StreamEndInfo {
return t.streamEndInfo
}
func (t *transStreamPublisher) TranscodeTracks() []*Track {
return t.allStreamTracks.All()
}
func (t *transStreamPublisher) LastStreamEndTime() time.Time {
return t.lastStreamEndTime
}
func (t *transStreamPublisher) SinkCount() int {
return t.sinkCount
}
func (t *transStreamPublisher) GetForwardTransStream() TransStream {
return t.forwardTransStream
}
func (t *transStreamPublisher) SetSourceID(id string) {
t.source = id
}
func NewTransStreamPublisher(source string) TransStreamPublisher {
return &transStreamPublisher{
transStreams: make(map[TransStreamID]TransStream),
transStreamSinks: make(map[TransStreamID]map[SinkID]Sink),
sinks: make(map[SinkID]Sink),
source: source,
}
}

View File

@@ -2,6 +2,7 @@ package stream
import (
"github.com/lkmio/avformat"
"github.com/lkmio/avformat/collections"
)
type Track struct {
@@ -9,8 +10,9 @@ type Track struct {
Pts int64 // 最新的PTS
Dts int64 // 最新的DTS
FrameDuration int // 单帧时长, timebase和推流一致
Packets collections.LinkedList[*collections.ReferenceCounter[*avformat.AVPacket]]
}
func NewTrack(stream *avformat.AVStream, dts, pts int64) *Track {
return &Track{stream, dts, pts, 0}
return &Track{stream, dts, pts, 0, collections.LinkedList[*collections.ReferenceCounter[*avformat.AVPacket]]{}}
}

108
web/demo.css Normal file
View File

@@ -0,0 +1,108 @@
.mainContainer {
display: block;
width: 100%;
margin-left: auto;
margin-right: auto;
}
@media screen and (min-width: 1152px) {
.mainContainer {
display: block;
width: 1152px;
margin-left: auto;
margin-right: auto;
}
}
.video-container {
position: relative;
margin-top: 8px;
}
.video-container:before {
display: block;
content: "";
width: 100%;
padding-bottom: 56.25%;
}
.video-container > div {
position: absolute;
top: 0;
left: 0;
right: 0;
bottom: 0;
}
.video-container video {
width: 100%;
height: 100%;
}
.urlInput {
display: block;
width: 100%;
margin-left: auto;
margin-right: auto;
margin-top: 8px;
margin-bottom: 8px;
}
.centeredVideo {
display: block;
width: 100%;
height: 100%;
margin-left: auto;
margin-right: auto;
margin-bottom: auto;
}
.controls {
display: block;
width: 100%;
text-align: left;
margin-left: auto;
margin-right: auto;
margin-top: 8px;
margin-bottom: 10px;
}
.logcatBox {
border-color: #CCCCCC;
font-size: 11px;
font-family: Menlo, Consolas, monospace;
display: block;
width: 100%;
text-align: left;
margin-left: auto;
margin-right: auto;
}
.url-input , .options {
font-size: 13px;
}
.url-input {
display: flex;
}
.url-input label {
flex: initial;
}
.url-input input {
flex: auto;
margin-left: 8px;
}
.url-input button {
flex: initial;
margin-left: 8px;
}
.options {
margin-top: 5px;
}
.hidden {
display: none;
}

227
web/index.html Normal file
View File

@@ -0,0 +1,227 @@
<!DOCTYPE html>
<html>
<head>
<meta content="text/html; charset=utf-8" http-equiv="Content-Type">
<title>flv.js demo</title>
<link rel="stylesheet" type="text/css" href="demo.css"/>
</head>
<body>
<div class="mainContainer">
<div>
<div id="streamURL">
<div class="url-input">
<label for="sURL">Stream URL:</label>
<input id="sURL" type="text" value="http://127.0.0.1/flv/7182741-1.flv"/>
<button onclick="switch_mds()">Switch to MediaDataSource</button>
</div>
<div class="options">
<input type="checkbox" id="isLive" onchange="saveSettings()"/>
<label for="isLive">isLive</label>
<input type="checkbox" id="withCredentials" onchange="saveSettings()"/>
<label for="withCredentials">withCredentials</label>
<input type="checkbox" id="hasAudio" onchange="saveSettings()" checked/>
<label for="hasAudio">hasAudio</label>
<input type="checkbox" id="hasVideo" onchange="saveSettings()" checked/>
<label for="hasVideo">hasVideo</label>
</div>
</div>
<div id="mediaSourceURL" class="hidden">
<div class="url-input">
<label for="msURL">MediaDataSource JsonURL:</label>
<input id="msURL" type="text" value="http://127.0.0.1/flv/7182741.json"/>
<button onclick="switch_url()">Switch to URL</button>
</div>
</div>
</div>
<div class="video-container">
<div>
<video name="videoElement" class="centeredVideo" controls autoplay>
Your browser is too old which doesn't support HTML5 video.
</video>
</div>
</div>
<div class="controls">
<button onclick="flv_load()">Load</button>
<button onclick="flv_start()">Start</button>
<button onclick="flv_pause()">Pause</button>
<button onclick="flv_destroy()">Destroy</button>
<input style="width:100px" type="text" name="seekpoint"/>
<button onclick="flv_seekto()">SeekTo</button>
</div>
<textarea name="logcatbox" class="logcatBox" rows="10" readonly></textarea>
</div>
<script src="https://cdnjs.cloudflare.com/ajax/libs/flv.js/1.6.2/flv.min.js"></script>
<script>
var checkBoxFields = ['isLive', 'withCredentials', 'hasAudio', 'hasVideo'];
var streamURL, mediaSourceURL;
function flv_load() {
console.log('isSupported: ' + flvjs.isSupported());
if (mediaSourceURL.className === '') {
var url = document.getElementById('msURL').value;
var xhr = new XMLHttpRequest();
xhr.open('GET', url, true);
xhr.onload = function (e) {
var mediaDataSource = JSON.parse(xhr.response);
flv_load_mds(mediaDataSource);
}
xhr.send();
} else {
var i;
var mediaDataSource = {
type: 'flv'
};
for (i = 0; i < checkBoxFields.length; i++) {
var field = checkBoxFields[i];
/** @type {HTMLInputElement} */
var checkbox = document.getElementById(field);
mediaDataSource[field] = checkbox.checked;
}
mediaDataSource['url'] = document.getElementById('sURL').value;
console.log('MediaDataSource', mediaDataSource);
flv_load_mds(mediaDataSource);
}
}
function flv_load_mds(mediaDataSource) {
var element = document.getElementsByName('videoElement')[0];
if (typeof player !== "undefined") {
if (player != null) {
player.unload();
player.detachMediaElement();
player.destroy();
player = null;
}
}
player = flvjs.createPlayer(mediaDataSource, {
enableWorker: false,
lazyLoadMaxDuration: 3 * 60,
bufferLength: 5, // 初始缓冲 5 秒
bufferTime: 10, // MediaSource 缓冲 10 秒
seekType: 'range',
});
player.attachMediaElement(element);
player.load();
}
function flv_start() {
player.play();
}
function flv_pause() {
player.pause();
}
function flv_destroy() {
player.pause();
player.unload();
player.detachMediaElement();
player.destroy();
player = null;
}
function flv_seekto() {
var input = document.getElementsByName('seekpoint')[0];
player.currentTime = parseFloat(input.value);
}
function switch_url() {
streamURL.className = '';
mediaSourceURL.className = 'hidden';
saveSettings();
}
function switch_mds() {
streamURL.className = 'hidden';
mediaSourceURL.className = '';
saveSettings();
}
function ls_get(key, def) {
try {
var ret = localStorage.getItem('flvjs_demo.' + key);
if (ret === null) {
ret = def;
}
return ret;
} catch (e) {}
return def;
}
function ls_set(key, value) {
try {
localStorage.setItem('flvjs_demo.' + key, value);
} catch (e) {}
}
function saveSettings() {
if (mediaSourceURL.className === '') {
ls_set('inputMode', 'MediaDataSource');
} else {
ls_set('inputMode', 'StreamURL');
}
var i;
for (i = 0; i < checkBoxFields.length; i++) {
var field = checkBoxFields[i];
/** @type {HTMLInputElement} */
var checkbox = document.getElementById(field);
ls_set(field, checkbox.checked ? '1' : '0');
}
var msURL = document.getElementById('msURL');
var sURL = document.getElementById('sURL');
ls_set('msURL', msURL.value);
ls_set('sURL', sURL.value);
console.log('save');
}
function loadSettings() {
var i;
for (i = 0; i < checkBoxFields.length; i++) {
var field = checkBoxFields[i];
/** @type {HTMLInputElement} */
var checkbox = document.getElementById(field);
var c = ls_get(field, checkbox.checked ? '1' : '0');
checkbox.checked = c === '1' ? true : false;
}
var msURL = document.getElementById('msURL');
var sURL = document.getElementById('sURL');
msURL.value = ls_get('msURL', msURL.value);
sURL.value = ls_get('sURL', sURL.value);
if (ls_get('inputMode', 'StreamURL') === 'StreamURL') {
switch_url();
} else {
switch_mds();
}
}
function showVersion() {
var version = flvjs.version;
document.title = document.title + " (v" + version + ")";
}
var logcatbox = document.getElementsByName('logcatbox')[0];
flvjs.LoggingControl.addLogListener(function(type, str) {
logcatbox.value = logcatbox.value + str + '\n';
logcatbox.scrollTop = logcatbox.scrollHeight;
});
document.addEventListener('DOMContentLoaded', function () {
streamURL = document.getElementById('streamURL');
mediaSourceURL = document.getElementById('mediaSourceURL');
loadSettings();
showVersion();
flv_load();
});
</script>
</body>
</html>

View File

@@ -38,7 +38,7 @@
<div style="margin-top: 10px;">
<div style="float: left">
<video id="videoview" width="310" autoplay muted controls ></video>
<video id="videoview" width="310" autoplay muted controls></video>
</div>
</div>
</body>
@@ -50,7 +50,7 @@
let remote_view = document.getElementById("videoview");
let source = document.getElementById("source").value;
let pc = new RTCPeerConnection(null);
// pc.addTransceiver("audio", {direction: "recvonly"});
// pc.addTransceiver("audio", {direction: "recvonly"});
pc.addTransceiver("video", {direction: "recvonly"});
let offer = await pc.createOffer();
@@ -80,7 +80,9 @@
}
console.log("offer:" + offer.sdp);
let url = source + ".rtc";
//source = generateRandomAlphanumeric10();
let url = window.location.origin + "/" + source + ".rtc";
fetch(url, {
method: 'POST',
body: JSON.stringify(data),