apps/plugin-rtmp
1
0
Fork 0
mirror of https://github.com/Monibuca/plugin-rtmp.git synced 2025-10-09 01:10:37 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

增加音视频同步输出功能

Browse Source
This commit is contained in:
langhuihui
2021-01-25 22:04:45 +08:00
parent 6084f8b96d
commit ad8cbf59e0
1 changed files with 52 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
netStream.go
View File

@@ -9,7 +9,6 @@ import (
"log" "log"
"net" "net"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
) )
@@ -148,6 +147,7 @@ func processRtmp(conn net.Conn) {
nalus = nalus[nalulen+nalulenSize:] nalus = nalus[nalulen+nalulenSize:]
} }
} }
close(stream.WaitPub)
} }
for { for {
if msg, err := nc.RecvMessage(); err == nil { if msg, err := nc.RecvMessage(); err == nil {
@@ -183,45 +183,23 @@ func processRtmp(conn net.Conn) {
pm := msg.MsgData.(*PlayMessage) pm := msg.MsgData.(*PlayMessage)
streamPath := nc.appName + "/" + strings.Split(pm.StreamName, "?")[0] streamPath := nc.appName + "/" + strings.Split(pm.StreamName, "?")[0]
nc.writeChunkSize = config.ChunkSize nc.writeChunkSize = config.ChunkSize
stream := engine.Subscriber{} var subscriber engine.Subscriber
stream.Type = "RTMP" subscriber.Type = "RTMP"
stream.ID = fmt.Sprintf("%s|%d", conn.RemoteAddr().String(), nc.streamID) subscriber.ID = fmt.Sprintf("%s|%d", conn.RemoteAddr().String(), nc.streamID)
if err = subscriber.Subscribe(streamPath); err == nil {
streams[nc.streamID] = &subscriber
err = nc.SendMessage(SEND_CHUNK_SIZE_MESSAGE, uint32(nc.writeChunkSize)) err = nc.SendMessage(SEND_CHUNK_SIZE_MESSAGE, uint32(nc.writeChunkSize))
err = nc.SendMessage(SEND_STREAM_IS_RECORDED_MESSAGE, nil) err = nc.SendMessage(SEND_STREAM_IS_RECORDED_MESSAGE, nil)
err = nc.SendMessage(SEND_STREAM_BEGIN_MESSAGE, nil) err = nc.SendMessage(SEND_STREAM_BEGIN_MESSAGE, nil)
err = nc.SendMessage(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Reset, Level_Status)) err = nc.SendMessage(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Reset, Level_Status))
err = nc.SendMessage(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Start, Level_Status)) err = nc.SendMessage(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Start, Level_Status))
if err == nil { vt, at := subscriber.VideoTracks[0], subscriber.AudioTracks[0]
streams[nc.streamID] = &stream
if err = stream.Subscribe(streamPath); err == nil {
vt, at := stream.VideoTracks[0], stream.AudioTracks[0]
err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.RtmpTag}) err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.RtmpTag})
if at.SoundFormat == 10 { if at.SoundFormat == 10 {
err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.RtmpTag}) err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.RtmpTag})
} }
var lastAudioTime, lastVideoTime uint32 var lastAudioTime, lastVideoTime uint32
var lock sync.Mutex go (&engine.TrackCP{at, vt}).Play(subscriber.Context, func(pack engine.AudioPack) {
go vt.Play(stream.Context, func(pack engine.VideoPack) {
if lastVideoTime == 0 {
lastVideoTime = pack.Timestamp
}
t := pack.Timestamp - lastVideoTime
lastVideoTime = pack.Timestamp
payload := utils.GetSlice(9 + len(pack.Payload))
defer utils.RecycleSlice(payload)
if pack.NalType == codec.NALU_IDR_Picture {
payload[0] = 0x17
} else {
payload[0] = 0x27
}
payload[1] = 0x01
utils.BigEndian.PutUint32(payload[5:], uint32(len(pack.Payload)))
copy(payload[9:], pack.Payload)
lock.Lock()
defer lock.Unlock()
err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: t, Payload: payload})
})
go at.Play(stream.Context, func(pack engine.AudioPack) {
if lastAudioTime == 0 { if lastAudioTime == 0 {
lastAudioTime = pack.Timestamp lastAudioTime = pack.Timestamp
} }
@@ -238,13 +216,25 @@ func processRtmp(conn net.Conn) {
payload[1] = 1 payload[1] = 1
} }
copy(payload[2:], pack.Payload) copy(payload[2:], pack.Payload)
lock.Lock()
defer lock.Unlock()
err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: t, Payload: payload}) err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: t, Payload: payload})
}) }, func(pack engine.VideoPack) {
if lastVideoTime == 0 {
lastVideoTime = pack.Timestamp
} }
t := pack.Timestamp - lastVideoTime
lastVideoTime = pack.Timestamp
payload := utils.GetSlice(9 + len(pack.Payload))
defer utils.RecycleSlice(payload)
if pack.NalType == codec.NALU_IDR_Picture {
payload[0] = 0x17
} else { } else {
return payload[0] = 0x27
}
payload[1] = 0x01
utils.BigEndian.PutUint32(payload[5:], uint32(len(pack.Payload)))
copy(payload[9:], pack.Payload)
err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: t, Payload: payload})
})
} }
case "closeStream": case "closeStream":
cm := msg.MsgData.(*CURDStreamMessage) cm := msg.MsgData.(*CURDStreamMessage)