适配引擎版本升级

go.mod

@@ -3,7 +3,7 @@ module github.com/Monibuca/plugin-rtmp/v3
 go 1.13
 
 require (
-	github.com/Monibuca/engine/v3 v3.0.0-alpha11
-	github.com/Monibuca/utils/v3 v3.0.0-alpha5
+	github.com/Monibuca/engine/v3 v3.0.0-beta5
+	github.com/Monibuca/utils/v3 v3.0.0-beta
 	github.com/logrusorgru/aurora v2.0.3+incompatible
 )

go.sum

@@ -2,8 +2,14 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Monibuca/engine/v3 v3.0.0-alpha11 h1:Z/ViugwmK+ECBK5MS9bUltrUU1UlcDjFTuJh4nxouK0=
 github.com/Monibuca/engine/v3 v3.0.0-alpha11/go.mod h1:eonu3UFn3W7NpHzSrACipxdAyOBCUwzlFUe1R7JjttE=
+github.com/Monibuca/engine/v3 v3.0.0-beta3 h1:/co+L2qCRZUq55S0LtYpY9xzOJiUUG3VGytYGFf1RD4=
+github.com/Monibuca/engine/v3 v3.0.0-beta3/go.mod h1:SMgnlwih4pBA/HkTLjKXZFYkv3ukRzFjv65CARRLVIk=
+github.com/Monibuca/engine/v3 v3.0.0-beta5 h1:b27ZQDfvf5dBMZbCSIUXItUwVIFs95fpkAV4xjN7BNE=
+github.com/Monibuca/engine/v3 v3.0.0-beta5/go.mod h1:SMgnlwih4pBA/HkTLjKXZFYkv3ukRzFjv65CARRLVIk=
 github.com/Monibuca/utils/v3 v3.0.0-alpha5 h1:IOyW/KJSRdRg+TPcgwkHLBynqfNQOV6p3iP7LgXEMFc=
 github.com/Monibuca/utils/v3 v3.0.0-alpha5/go.mod h1:3xYmhQbgAZBHLyIMteUCd1va+1z/xnd72B585mCaT3c=
+github.com/Monibuca/utils/v3 v3.0.0-beta h1:z4p/BSH5J9Ja/gwoDmj1RyN+b0q28Nmn/fqXiwq2hGY=
+github.com/Monibuca/utils/v3 v3.0.0-beta/go.mod h1:mQYP/OMox1tkWP6Qut7pBfARr1TXSRkK662dexQl6kI=
 github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478 h1:Db9StoJ6RZN3YttC0Pm0I4Y5izITRYch3RMbT59BYN0=
 github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478/go.mod h1:0j1+svBH8ABEIPdUP0AIg4qedsybnXGJBakCEw8cfoo=
 github.com/funny/utest v0.0.0-20161029064919-43870a374500 h1:Z0r1CZnoIWFB/Uiwh1BU5FYmuFe6L5NPi6XWQEmsTRg=
@@ -18,8 +24,12 @@ github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
 github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
 github.com/pion/rtp v1.6.2 h1:iGBerLX6JiDjB9NXuaPzHyxHFG9JsIEdgwTC0lp5n/U=
 github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
+github.com/pion/rtp v1.6.5 h1:o2cZf8OascA5HF/b0PAbTxRKvOWxTQxWYt7SlToxFGI=
+github.com/pion/rtp v1.6.5/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/q191201771/naza v0.19.1 h1:4KLcxT2CHztO+7miPRtBG3FFgadSQYQw1gPPPKN7rnY=
+github.com/q191201771/naza v0.19.1/go.mod h1:5LeGupZZFtYP1g/S203n9vXoUNVdlRnPIfM6rExjqt0=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

netStream.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/Monibuca/engine/v3"
 	"github.com/Monibuca/utils/v3"
-	"github.com/Monibuca/utils/v3/codec"
 )
 
 func ListenRtmp(addr string) error {
@@ -81,97 +80,7 @@ func processRtmp(conn net.Conn) {
 		return
 	}
 	var rec_audio, rec_video func(*Chunk)
-	var abslouteTs uint32
-	rec_audio = func(msg *Chunk) {
-		va := engine.NewAudioTrack()
-		tmp := msg.Body[0]
-		soundFormat := tmp >> 4
-		switch soundFormat {
-		case 10:
-			if msg.Body[1] != 0 {
-				return
-			}
-			va.SoundFormat = soundFormat
-			config1, config2 := msg.Body[2], msg.Body[3]
-			//audioObjectType = (config1 & 0xF8) >> 3
-			// 1 AAC MAIN 	ISO/IEC 14496-3 subpart 4
-			// 2 AAC LC 	ISO/IEC 14496-3 subpart 4
-			// 3 AAC SSR 	ISO/IEC 14496-3 subpart 4
-			// 4 AAC LTP 	ISO/IEC 14496-3 subpart 4
-			va.SoundRate = codec.SamplingFrequencies[((config1&0x7)<<1)|(config2>>7)]
-			va.Channels = ((config2 >> 3) & 0x0F) //声道
-			//frameLengthFlag = (config2 >> 2) & 0x01
-			//dependsOnCoreCoder = (config2 >> 1) & 0x01
-			//extensionFlag = config2 & 0x01
-			va.RtmpTag = msg.Body
-			rec_audio = func(msg *Chunk) {
-				if msg.Timestamp == 0xffffff {
-					abslouteTs += msg.ExtendTimestamp
-				} else {
-					abslouteTs += msg.Timestamp // 绝对时间戳
-				}
-				va.Push(abslouteTs, msg.Body[2:])
-			}
-			stream.SetOriginAT(va)
-			return
-		case 7, 8:
-			va.RtmpTag = msg.Body
-			va.SoundFormat = soundFormat
-			va.SoundRate = codec.SoundRate[(tmp&0x0c)>>2] // 采样率 0 = 5.5 kHz or 1 = 11 kHz or 2 = 22 kHz or 3 = 44 kHz
-			va.SoundSize = (tmp & 0x02) >> 1              // 采样精度 0 = 8-bit samples or 1 = 16-bit samples
-			va.Channels = tmp&0x01 + 1                    // 0 单声道，1立体声
-			rec_audio = func(msg *Chunk) {
-				if msg.Timestamp == 0xffffff {
-					abslouteTs += msg.ExtendTimestamp
-				} else {
-					abslouteTs += msg.Timestamp // 绝对时间戳
-				}
-				va.Push(abslouteTs, msg.Body[1:])
-			}
-			stream.SetOriginAT(va)
-		}
-	}
-	rec_video = func(msg *Chunk) {
-		codecId := msg.Body[0] & 0x0F
-		// 等待AVC序列帧
-		if codecId != 7 && codecId != 12 || msg.Body[1] != 0 {
-			return
-		}
-		vt := engine.NewVideoTrack()
-		vt.CodecID = codecId
-		vt.RtmpTag = msg.Body
-		var info codec.AVCDecoderConfigurationRecord
-		//0:codec,1:IsAVCSequence,2~4:compositionTime
-		if _, err := info.Unmarshal(msg.Body[5:]); err == nil {
-			var pack engine.VideoPack
-			pack.Payload = info.SequenceParameterSetNALUnit
-			vt.Push(pack)
-			pack.Payload = info.PictureParameterSetNALUnit
-			vt.Push(pack)
-		}
-		nalulenSize := int(info.LengthSizeMinusOne&3 + 1)
-		stream.SetOriginVT(vt)
-		rec_video = func(msg *Chunk) {
-			var pack engine.VideoPack
-			pack.CompositionTime = utils.BigEndian.Uint24(msg.Body[2:5])
-			nalus := msg.Body[5:]
-			if msg.Timestamp == 0xffffff {
-				abslouteTs += msg.ExtendTimestamp
-			} else {
-				abslouteTs += msg.Timestamp // 绝对时间戳
-			}
-			pack.Timestamp = abslouteTs
-			for len(nalus) > nalulenSize {
-				nalulen := 0
-				for i := 0; i < nalulenSize; i++ {
-					nalulen += int(nalus[i]) << (8 * (nalulenSize - i - 1))
-				}
-				pack.Payload = nalus[nalulenSize : nalulen+nalulenSize]
-				vt.Push(pack)
-				nalus = nalus[nalulen+nalulenSize:]
-			}
-		}
-	}
+
 	for {
 		if msg, err := nc.RecvMessage(); err == nil {
 			if msg.MessageLength <= 0 {
@@ -194,9 +103,26 @@ func processRtmp(conn net.Conn) {
 				case "publish":
 					pm := msg.MsgData.(*PublishMessage)
 					streamPath := nc.appName + "/" + strings.Split(pm.PublishingName, "?")[0]
-					if pub := new(engine.Publisher); pub.Publish(streamPath) {
-						pub.Type = "RTMP"
-						stream = pub.Stream
+					stream = &engine.Stream{Type: "RTMP", StreamPath: streamPath}
+					if stream.Publish() {
+						var abslouteTs uint32
+						vt, at := stream.NewVideoTrack(0), stream.NewAudioTrack(0)
+						rec_audio = func(msg *Chunk) {
+							if msg.Timestamp == 0xffffff {
+								abslouteTs += msg.ExtendTimestamp
+							} else {
+								abslouteTs += msg.Timestamp // 绝对时间戳
+							}
+							at.PushByteStream(engine.AudioPack{Timestamp: abslouteTs, Payload: msg.Body})
+						}
+						rec_video = func(msg *Chunk) {
+							if msg.Timestamp == 0xffffff {
+								abslouteTs += msg.ExtendTimestamp
+							} else {
+								abslouteTs += msg.Timestamp // 绝对时间戳
+							}
+							vt.PushByteStream(engine.VideoPack{Timestamp: abslouteTs, Payload: msg.Body})
+						}
 						err = nc.SendMessage(SEND_STREAM_BEGIN_MESSAGE, nil)
 						err = nc.SendMessage(SEND_PUBLISH_START_MESSAGE, newPublishResponseMessageData(nc.streamID, NetStream_Publish_Start, Level_Status))
 					} else {
@@ -209,6 +135,7 @@ func processRtmp(conn net.Conn) {
 					subscriber := engine.Subscriber{
 						Type:             "RTMP",
 						ID:               fmt.Sprintf("%s|%d", conn.RemoteAddr().String(), nc.streamID),
+						ByteStreamFormat: true,
 					}
 					if err = subscriber.Subscribe(streamPath); err == nil {
 						streams[nc.streamID] = &subscriber
@@ -217,38 +144,34 @@ func processRtmp(conn net.Conn) {
 						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_Start, Level_Status))
-						vt, at := subscriber.OriginVideoTrack, subscriber.OriginAudioTrack
+						vt, at := subscriber.WaitVideoTrack(), subscriber.WaitAudioTrack()
 						var lastTimeStamp uint32
-						getDeltaTime := func(ts uint32) (t uint32) {
+						var getDeltaTime func(uint32) uint32
+						getDeltaTime = func(ts uint32) (t uint32) {
+							lastTimeStamp = ts
+							getDeltaTime = func(ts uint32) (t uint32) {
 								t = ts - lastTimeStamp
 								lastTimeStamp = ts
 								return
 							}
+							return
+						}
 						if vt != nil {
-							err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.RtmpTag})
+							err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.ExtraData.Payload})
 							subscriber.OnVideo = func(pack engine.VideoPack) {
-								payload := pack.ToRTMPTag()
-								defer utils.RecycleSlice(payload)
-								lastTimeStamp = pack.Timestamp
-								err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Timestamp: 0, Payload: payload})
+								err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Timestamp: 0, Payload: pack.Payload})
 								subscriber.OnVideo = func(pack engine.VideoPack) {
-									payload := pack.ToRTMPTag()
-									defer utils.RecycleSlice(payload)
-									err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: getDeltaTime(pack.Timestamp), Payload: payload})
+									err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: getDeltaTime(pack.Timestamp), Payload: pack.Payload})
 								}
 							}
 						}
 						if at != nil {
 							subscriber.OnAudio = func(pack engine.AudioPack) {
-								payload := pack.ToRTMPTag(at.RtmpTag[0])
-								defer utils.RecycleSlice(payload)
-								if at.SoundFormat == 10 {
-									err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.RtmpTag})
+								if at.CodecID == 10 {
+									err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.ExtraData})
 								}
 								subscriber.OnAudio = func(pack engine.AudioPack) {
-									payload := pack.ToRTMPTag(at.RtmpTag[0])
-									defer utils.RecycleSlice(payload)
-									err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: getDeltaTime(pack.Timestamp), Payload: payload})
+									err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: getDeltaTime(pack.Timestamp), Payload: pack.Payload})
 								}
 								subscriber.OnAudio(pack)
 							}