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

初步调通

Browse Source
This commit is contained in:
dexter
2022-02-06 08:50:41 +08:00
parent a5e5d31fda
commit c1fedbc529
10 changed files with 608 additions and 1093 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
.idea

376
amf.go
View File

@@ -1,14 +1,10 @@
package rtmp
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"log"
"reflect"
"github.com/Monibuca/utils/v3"
"github.com/Monibuca/engine/v4/util"
)
// Action Message Format -- AMF 0
@@ -77,93 +73,23 @@ const (
var END_OBJ = []byte{0, 0, AMF0_END_OBJECT}
type AMFObject interface{}
type AMFValue any
type AMFObjects map[string]AMFObject
func newAMFObjects() AMFObjects {
return make(AMFObjects, 0)
}
func DecodeAMFObject(obj interface{}, key string) interface{} {
if v, ok := obj.(AMFObjects)[key]; ok {
return v
}
return nil
}
type AMFObject map[string]AMFValue
type AMF struct {
*bytes.Buffer
util.Buffer
}
func newAMFEncoder() *AMF {
return &AMF{
new(bytes.Buffer),
}
}
var ObjectEnd = &struct{}{}
var Undefined = &struct{}{}
func newAMFDecoder(b []byte) *AMF {
return &AMF{
bytes.NewBuffer(b),
}
}
func (amf *AMF) readSize() (int, error) {
b, err := readBytes(amf.Buffer, 4)
size := int(utils.BigEndian.Uint32(b))
utils.RecycleSlice(b)
return size, err
}
func (amf *AMF) readSize16() (int, error) {
b, err := readBytes(amf.Buffer, 2)
size := int(utils.BigEndian.Uint16(b))
utils.RecycleSlice(b)
return size, err
}
func (amf *AMF) readObjects() (obj []AMFObject, err error) {
obj = make([]AMFObject, 0)
for amf.Len() > 0 {
if v, err := amf.decodeObject(); err == nil {
obj = append(obj, v)
} else {
return obj, err
}
}
return
}
func (amf *AMF) writeObjects(obj []AMFObject) (err error) {
for _, v := range obj {
switch data := v.(type) {
case string:
err = amf.writeString(data)
case float64:
err = amf.writeNumber(data)
case bool:
err = amf.writeBool(data)
case AMFObjects:
err = amf.encodeObject(data)
case nil:
err = amf.writeNull()
default:
log.Printf("amf encode unknown type:%v", reflect.TypeOf(data).Name())
}
}
return
}
func (amf *AMF) decodeObject() (obj AMFObject, err error) {
func (amf *AMF) decodeObject() (obj AMFValue) {
if amf.Len() == 0 {
return nil, errors.New(fmt.Sprintf("no enough bytes, %v/%v", amf.Len(), 1))
}
var t byte
if t, err = amf.ReadByte(); err != nil {
return
}
if err = amf.UnreadByte(); err != nil {
return
fmt.Sprintf("no enough bytes, %v/%v", amf.Len(), 1)
return nil
}
t := amf.Buffer[0]
switch t {
case AMF0_NUMBER:
return amf.readNumber()
@@ -178,255 +104,183 @@ func (amf *AMF) decodeObject() (obj AMFObject, err error) {
case AMF0_NULL:
return amf.readNull()
case AMF0_UNDEFINED:
_, err = amf.ReadByte()
return "Undefined", err
amf.ReadByte()
return Undefined
case AMF0_REFERENCE:
log.Println("reference-type.(AMF0_REFERENCE)")
case AMF0_ECMA_ARRAY:
return amf.readECMAArray()
case AMF0_END_OBJECT:
_, err = amf.ReadByte()
return "ObjectEnd", err
amf.ReadByte()
return ObjectEnd
case AMF0_STRICT_ARRAY:
return amf.readStrictArray()
case AMF0_DATE:
return amf.readDate()
case AMF0_LONG_STRING:
case AMF0_LONG_STRING,
AMF0_XML_DOCUMENT:
return amf.readLongString()
case AMF0_UNSUPPORTED:
log.Println("If a type cannot be serialized a special unsupported marker can be used in place of the type.(AMF0_UNSUPPORTED)")
case AMF0_RECORDSET:
log.Println("This type is not supported and is reserved for future use.(AMF0_RECORDSET)")
case AMF0_XML_DOCUMENT:
return amf.readLongString()
case AMF0_TYPED_OBJECT:
log.Println("If a strongly typed object has an alias registered for its class then the type name will also be serialized. Typed objects are considered complex types and reoccurring instances can be sent by reference.(AMF0_TYPED_OBJECT)")
case AMF0_AVMPLUS_OBJECT:
log.Println("AMF0_AVMPLUS_OBJECT")
default:
log.Println("Unkonw type.")
fmt.Sprintf("Unsupported type %v", t)
}
return nil, errors.New(fmt.Sprintf("Unsupported type %v", t))
return nil
}
func (amf *AMF) encodeObject(t AMFObjects) (err error) {
err = amf.writeObject()
defer amf.writeObjectEnd()
func (amf *AMF) writeObject(t AMFObject) {
if t == nil {
return
}
amf.Malloc(1)[0] = AMF0_OBJECT
defer amf.Write(END_OBJ)
for k, vv := range t {
switch vvv := vv.(type) {
case string:
if err = amf.writeObjectString(k, vvv); err != nil {
return
}
amf.writeObjectString(k, vvv)
case float64, uint, float32, int, int16, int32, int64, uint16, uint32, uint64, uint8, int8:
if err = amf.writeObjectNumber(k, utils.ToFloat64(vvv)); err != nil {
return
}
amf.writeObjectNumber(k, util.ToFloat64(vv))
case bool:
if err = amf.writeObjectBool(k, vvv); err != nil {
return
}
}
}
return
}
func (amf *AMF) readDate() (t uint64, err error) {
_, err = amf.ReadByte() // 取出第一个字节 8 Bit == 1 Byte. buf - 1.
var b []byte
b, err = readBytes(amf.Buffer, 8) // 在取出8个字节,并且读到b中. buf - 8
t = utils.BigEndian.Uint64(b)
utils.RecycleSlice(b)
b, err = readBytes(amf.Buffer, 2)
utils.RecycleSlice(b)
return t, err
}
func (amf *AMF) readStrictArray() (list []AMFObject, err error) {
list = make([]AMFObject, 0)
_, err = amf.ReadByte()
var size int
size, err = amf.readSize()
for i := 0; i < size; i++ {
if obj, err := amf.decodeObject(); err != nil {
return list, err
} else {
list = append(list, obj)
amf.writeObjectBool(k, vvv)
}
}
return
}
func (amf *AMF) readECMAArray() (m AMFObjects, err error) {
m = make(AMFObjects, 0)
_, err = amf.ReadByte()
var size int
size, err = amf.readSize()
for i := 0; i < size; i++ {
var k string
var v AMFObject
if k, err = amf.readString1(); err == nil {
if v, err = amf.decodeObject(); err == nil {
if k != "" || "ObjectEnd" != v {
func (amf *AMF) readDate() uint64 {
amf.ReadByte()
ret := amf.ReadUint64()
amf.ReadN(2) // timezone
return ret
}
func (amf *AMF) readStrictArray() (list []AMFValue) {
amf.ReadByte()
size := amf.ReadUint16()
for i := uint16(0); i < size; i++ {
list = append(list, amf.decodeObject())
}
return
}
func (amf *AMF) readECMAArray() (m AMFObject) {
m = make(AMFObject, 0)
amf.ReadByte()
size := amf.ReadUint16()
for i := uint16(0); i < size; i++ {
k := amf.readString1()
v := amf.decodeObject()
if k == "" && v == ObjectEnd {
return
}
m[k] = v
continue
}
}
}
return
}
return
}
func (amf *AMF) readString() (str string, err error) {
_, err = amf.ReadByte() // 取出第一个字节 8 Bit == 1 Byte. buf - 1.
func (amf *AMF) readString() string {
if amf.Len() == 0 {
return ""
}
amf.ReadByte()
return amf.readString1()
}
func (amf *AMF) readString1() (str string, err error) {
var size int
size, err = amf.readSize16()
var b []byte
b, err = readBytes(amf.Buffer, size) // 读取全部数据,读取长度为l,因为这两个字节(l变量)保存的是数据长度
str = string(b)
utils.RecycleSlice(b)
return
func (amf *AMF) readString1() string {
return string(amf.ReadN(int(amf.ReadUint16())))
}
func (amf *AMF) readLongString() (str string, err error) {
_, err = amf.ReadByte()
var size int
size, err = amf.readSize()
var b []byte
b, err = readBytes(amf.Buffer, size) // 读取全部数据,读取长度为l,因为这两个字节(l变量)保存的是数据长度
str = string(b)
utils.RecycleSlice(b)
return
func (amf *AMF) readLongString() string {
amf.ReadByte()
return string(amf.ReadN(int(amf.ReadUint32())))
}
func (amf *AMF) readNull() (AMFObject, error) {
_, err := amf.ReadByte()
return nil, err
func (amf *AMF) readNull() any {
amf.ReadByte()
return nil
}
func (amf *AMF) readNumber() (num float64, err error) {
// binary.read 会读取8个字节(float64),如果小于8个字节返回一个`io.ErrUnexpectedEOF`,如果大于就会返回`io.ErrShortBuffer`,读取完毕会有`io.EOF`
_, err = amf.ReadByte()
err = binary.Read(amf, binary.BigEndian, &num)
return num, err
func (amf *AMF) readNumber() (num float64) {
if amf.Len() == 0 {
return 0
}
amf.ReadByte()
return amf.ReadFloat64()
}
func (amf *AMF) readBool() (f bool, err error) {
_, err = amf.ReadByte()
if b, err := amf.ReadByte(); err == nil {
return b == 1, nil
func (amf *AMF) readBool() bool {
if amf.Len() == 0 {
return false
}
return
amf.ReadByte()
return amf.ReadByte() == 1
}
func (amf *AMF) readObject() (m AMFObjects, err error) {
_, err = amf.ReadByte()
m = make(AMFObjects, 0)
var k string
var v AMFObject
func (amf *AMF) readObject() (m AMFObject) {
if amf.Len() == 0 {
return nil
}
amf.ReadByte()
m = make(AMFObject, 0)
for {
if k, err = amf.readString1(); err != nil {
break
}
if v, err = amf.decodeObject(); err != nil {
break
}
if k == "" && "ObjectEnd" == v {
break
k := amf.readString1()
v := amf.decodeObject()
if k == "" && v == ObjectEnd {
return
}
m[k] = v
}
return m, err
}
func readBytes(buf *bytes.Buffer, length int) (b []byte, err error) {
b = utils.GetSlice(length)
if i, _ := buf.Read(b); length != i {
err = errors.New(fmt.Sprintf("not enough bytes,%v/%v", buf.Len(), length))
}
return
}
func (amf *AMF) writeSize16(l int) (err error) {
b := utils.GetSlice(2)
defer utils.RecycleSlice(b)
utils.BigEndian.PutUint16(b, uint16(l))
_, err = amf.Write(b)
return
}
func (amf *AMF) writeString(value string) error {
v := []byte(value)
err := amf.WriteByte(byte(AMF0_STRING))
if err != nil {
return err
}
if err = amf.writeSize16(len(v)); err != nil {
return err
}
_, err = amf.Write(v)
return err
func (amf *AMF) writeSize16(l int) {
util.PutBE(amf.Malloc(2), l)
}
func (amf *AMF) writeNull() error {
return amf.WriteByte(byte(AMF0_NULL))
func (amf *AMF) writeString(value string) {
amf.WriteUint8(AMF0_STRING)
amf.writeSize16(len(value))
amf.WriteString(value)
}
func (amf *AMF) writeBool(b bool) error {
if err := amf.WriteByte(byte(AMF0_BOOLEAN)); err != nil {
return err
func (amf *AMF) writeNull() {
amf.WriteUint8(AMF0_NULL)
}
func (amf *AMF) writeBool(b bool) {
amf.WriteUint8(AMF0_BOOLEAN)
if b {
return amf.WriteByte(byte(1))
amf.WriteUint8(1)
}
return amf.WriteByte(byte(0))
amf.WriteUint8(0)
}
func (amf *AMF) writeNumber(b float64) error {
if err := amf.WriteByte(byte(AMF0_NUMBER)); err != nil {
return err
}
return binary.Write(amf, binary.BigEndian, b)
func (amf *AMF) writeNumber(b float64) {
amf.WriteUint8(AMF0_NUMBER)
amf.WriteFloat64(b)
}
func (amf *AMF) writeObject() error {
return amf.WriteByte(byte(AMF0_OBJECT))
}
func (amf *AMF) writeKey(key string) (err error) {
keyByte := []byte(key)
if err = amf.writeSize16(len(keyByte)); err != nil {
return
}
if _, err = amf.Write(keyByte); err != nil {
return
}
return
}
func (amf *AMF) writeObjectString(key, value string) error {
if err := amf.writeKey(key); err != nil {
return err
}
return amf.writeString(value)
func (amf *AMF) writeKey(key string) {
amf.writeSize16(len(key))
amf.WriteString(key)
}
func (amf *AMF) writeObjectBool(key string, f bool) error {
if err := amf.writeKey(key); err != nil {
return err
}
return amf.writeBool(f)
func (amf *AMF) writeObjectString(key, value string) {
amf.writeKey(key)
amf.writeString(value)
}
func (amf *AMF) writeObjectNumber(key string, value float64) error {
if err := amf.writeKey(key); err != nil {
return err
}
return amf.writeNumber(value)
func (amf *AMF) writeObjectBool(key string, f bool) {
amf.writeKey(key)
amf.writeBool(f)
}
func (amf *AMF) writeObjectEnd() error {
_, err := amf.Write(END_OBJ)
return err
func (amf *AMF) writeObjectNumber(key string, value float64) {
amf.writeKey(key)
amf.writeNumber(value)
}

129
chunk.go
View File

@@ -1,9 +1,9 @@
package rtmp
import (
"errors"
"encoding/binary"
"github.com/Monibuca/utils/v3"
"github.com/Monibuca/engine/v4/util"
)
// RTMP协议中基本的数据单元称为消息(Message).
@@ -26,9 +26,9 @@ const (
)
type Chunk struct {
*ChunkHeader
ChunkHeader
Body []byte
MsgData interface{}
MsgData RtmpMessage
}
func (c *Chunk) Encode(msg RtmpMessage) {
@@ -36,14 +36,16 @@ func (c *Chunk) Encode(msg RtmpMessage) {
c.Body = msg.Encode()
c.MessageLength = uint32(len(c.Body))
}
func (c *Chunk) Recycle() {
chunkMsgPool.Put(c)
}
type ChunkHeader struct {
ChunkBasicHeader
ChunkMessageHeader
ChunkExtendedTimestamp
// Extended Timestamp (0 or 4 bytes): This field is present in certain
// circumstances depending on the encoded timestamp or timestamp
// delta field in the Chunk Message header. See Section 5.3.1.3 for
// more information
ExtendTimestamp uint32 `json:",omitempty"` // 标识该字段的数据可忽略
}
// Basic Header (1 to 3 bytes) : This field encodes the chunk stream ID
@@ -66,14 +68,6 @@ type ChunkMessageHeader struct {
MessageStreamID uint32 `json:""` // 4 byte
}
// Extended Timestamp (0 or 4 bytes): This field is present in certain
// circumstances depending on the encoded timestamp or timestamp
// delta field in the Chunk Message header. See Section 5.3.1.3 for
// more information
type ChunkExtendedTimestamp struct {
ExtendTimestamp uint32 `json:",omitempty"` // 标识该字段的数据可忽略
}
// ChunkBasicHeader会决定ChunkMessgaeHeader,ChunkMessgaeHeader有4种(0,3,7,11 Bytes),因此可能有4种头.
// 1 -> ChunkBasicHeader(1) + ChunkMessageHeader(0)
@@ -81,76 +75,49 @@ type ChunkExtendedTimestamp struct {
// 8 -> ChunkBasicHeader(1) + ChunkMessageHeader(7)
// 12 -> ChunkBasicHeader(1) + ChunkMessageHeader(11)
func (nc *NetConnection) encodeChunk12(head *ChunkHeader, payload []byte) (need []byte, err error) {
b := utils.GetSlice(12)
//chunkBasicHead
b[0] = byte(RTMP_CHUNK_HEAD_12 + head.ChunkBasicHeader.ChunkStreamID)
utils.BigEndian.PutUint24(b[1:], head.ChunkMessageHeader.Timestamp)
utils.BigEndian.PutUint24(b[4:], head.ChunkMessageHeader.MessageLength)
b[7] = head.ChunkMessageHeader.MessageTypeID
utils.LittleEndian.PutUint32(b[8:], uint32(head.ChunkMessageHeader.MessageStreamID))
nc.Write(b)
utils.RecycleSlice(b)
nc.writeSeqNum += 12
func (nc *NetConnection) encodeChunk12(head *ChunkHeader) []byte {
b := util.Buffer(make([]byte, 0, 16))
b.WriteUint8(byte(RTMP_CHUNK_HEAD_12 + head.ChunkStreamID))
b.WriteUint24(head.Timestamp)
b.WriteUint24(head.MessageLength)
b.WriteUint8(head.MessageTypeID)
binary.LittleEndian.PutUint32(b.Malloc(4), head.MessageStreamID)
if head.ChunkMessageHeader.Timestamp == 0xffffff {
b := utils.GetSlice(4)
utils.LittleEndian.PutUint32(b, head.ChunkExtendedTimestamp.ExtendTimestamp)
nc.Write(b)
utils.RecycleSlice(b)
nc.writeSeqNum += 4
binary.LittleEndian.PutUint32(b.Malloc(4), head.ExtendTimestamp)
}
return nc.writeChunk(payload)
return b
}
func (nc *NetConnection) encodeChunk8(head *ChunkHeader, payload []byte) (need []byte, err error) {
b := utils.GetSlice(8)
//chunkBasicHead
b[0] = byte(RTMP_CHUNK_HEAD_8 + head.ChunkBasicHeader.ChunkStreamID)
utils.BigEndian.PutUint24(b[1:], head.ChunkMessageHeader.Timestamp)
utils.BigEndian.PutUint24(b[4:], head.ChunkMessageHeader.MessageLength)
b[7] = head.ChunkMessageHeader.MessageTypeID
nc.Write(b)
utils.RecycleSlice(b)
nc.writeSeqNum += 8
return nc.writeChunk(payload)
func (nc *NetConnection) encodeChunk8(head *ChunkHeader) []byte {
b := util.Buffer(make([]byte, 0, 8))
b.WriteUint8(byte(RTMP_CHUNK_HEAD_8 + head.ChunkStreamID))
b.WriteUint24(head.Timestamp)
b.WriteUint24(head.MessageLength)
b.WriteUint8(head.MessageTypeID)
return b
}
func (nc *NetConnection) encodeChunk4(head *ChunkHeader, payload []byte, size int) (need []byte, err error) {
if size > RTMP_MAX_CHUNK_SIZE || payload == nil || len(payload) == 0 {
return nil, errors.New("chunk error")
}
b := utils.GetSlice(4)
//chunkBasicHead
b[0] = byte(RTMP_CHUNK_HEAD_4 + head.ChunkBasicHeader.ChunkStreamID)
utils.BigEndian.PutUint24(b[1:], head.ChunkMessageHeader.Timestamp)
nc.Write(b)
utils.RecycleSlice(b)
nc.writeSeqNum += 4
if len(payload) > size {
nc.Write(payload[0:size])
nc.writeSeqNum += uint32(size)
need = payload[size:]
} else {
nc.Write(payload)
nc.writeSeqNum += uint32(len(payload))
}
return
}
// func (nc *NetConnection) encodeChunk4(head *ChunkHeader, payload []byte, size int) (need []byte, err error) {
// if size > RTMP_MAX_CHUNK_SIZE || payload == nil || len(payload) == 0 {
// return nil, errors.New("chunk error")
// }
// b := make([]byte, 4)
// //chunkBasicHead
// b[0] = byte(RTMP_CHUNK_HEAD_4 + head.ChunkStreamID)
// util.PutBE(b[1:4], head.Timestamp)
// nc.Write(b)
// nc.writeSeqNum += 4
// if len(payload) > size {
// nc.Write(payload[0:size])
// nc.writeSeqNum += uint32(size)
// need = payload[size:]
// } else {
// nc.Write(payload)
// nc.writeSeqNum += uint32(len(payload))
// }
// return
// }
func (nc *NetConnection) encodeChunk1(head *ChunkHeader, payload []byte) (need []byte, err error) {
err = nc.WriteByte(byte(RTMP_CHUNK_HEAD_1 + head.ChunkBasicHeader.ChunkStreamID))
nc.writeSeqNum++
return nc.writeChunk(payload)
}
func (nc *NetConnection) writeChunk(payload []byte) (need []byte, err error) {
if payloadLen := len(payload); payloadLen > nc.writeChunkSize {
_, err = nc.Write(payload[:nc.writeChunkSize])
nc.writeSeqNum += uint32(nc.writeChunkSize)
need = payload[nc.writeChunkSize:]
} else {
_, err = nc.Write(payload)
nc.writeSeqNum += uint32(payloadLen)
}
return
func (nc *NetConnection) encodeChunk1(head *ChunkHeader) []byte {
return []byte{byte(RTMP_CHUNK_HEAD_1 + head.ChunkStreamID)}
}

10
go.mod
View File

@@ -1,9 +1,5 @@
module github.com/Monibuca/plugin-rtmp/v3
module github.com/Monibuca/plugin-rtmp/v4
go 1.13
go 1.18
require (
github.com/Monibuca/engine/v3 v3.4.5
github.com/Monibuca/utils/v3 v3.0.5
github.com/logrusorgru/aurora v2.0.3+incompatible
)
require github.com/logrusorgru/aurora v2.0.3+incompatible

79
go.sum
View File

@@ -1,81 +1,2 @@
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/BurntSushi/toml v0.4.1 h1:GaI7EiDXDRfa8VshkTj7Fym7ha+y8/XxIgD2okUIjLw=
github.com/BurntSushi/toml v0.4.1/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
github.com/Monibuca/engine/v3 v3.4.5 h1:gPRsliBVC70EhKK7sJz/3LoNO7lfnXNC8uUaOIHZkUE=
github.com/Monibuca/engine/v3 v3.4.5/go.mod h1:Dik9pFxU9TFI5vj8Sv5QXZM+ooCs2fm9P7Uhe4yYNkQ=
github.com/Monibuca/utils/v3 v3.0.5 h1:w14x0HkWTbF4MmHbINLlOwe4VJNoSOeaQChMk5E/4es=
github.com/Monibuca/utils/v3 v3.0.5/go.mod h1:RpNS95gapWs6gimwh8Xn2x72FN5tO7Powabj7dTFyvE=
github.com/cnotch/apirouter v0.0.0-20200731232942-89e243a791f3/go.mod h1:5deJPLON/x/s2dLOQfuKS0lenhOIT4xX0pvtN/OEIuY=
github.com/cnotch/ipchub v1.1.0 h1:hH0lh2mU3AZXPiqMwA0pdtqrwo7PFIMRGush9OobMUs=
github.com/cnotch/ipchub v1.1.0/go.mod h1:2PbeBs2q2VxxTVCn1eYCDwpAWuVXbq1+N0FU7GimOH4=
github.com/cnotch/loader v0.0.0-20200405015128-d9d964d09439/go.mod h1:oWpDagHB6p+Kqqq7RoRZKyC4XAXft50hR8pbTxdbYYs=
github.com/cnotch/queue v0.0.0-20200326024423-6e88bdbf2ad4/go.mod h1:zOssjAlNusOxvtaqT+EMA+Iyi8rrtKr4/XfzN1Fgoeg=
github.com/cnotch/queue v0.0.0-20201224060551-4191569ce8f6/go.mod h1:zOssjAlNusOxvtaqT+EMA+Iyi8rrtKr4/XfzN1Fgoeg=
github.com/cnotch/scheduler v0.0.0-20200522024700-1d2da93eefc5/go.mod h1:F4GE3SZkJZ8an1Y0ZCqvSM3jeozNuKzoC67erG1PhIo=
github.com/cnotch/xlog v0.0.0-20201208005456-cfda439cd3a0/go.mod h1:RW9oHsR79ffl3sR3yMGgxYupMn2btzdtJUwoxFPUE5E=
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/emitter-io/address v1.0.0/go.mod h1:GfZb5+S/o8694B1GMGK2imUYQyn2skszMvGNA5D84Ug=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
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=
github.com/funny/utest v0.0.0-20161029064919-43870a374500/go.mod h1:mUn39tBov9jKnTWV1RlOYoNzxdBFHiSzXWdY1FoNGGg=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
github.com/kelindar/process v0.0.0-20170730150328-69a29e249ec3/go.mod h1:+lTCLnZFXOkqwD8sLPl6u4erAc0cP8wFegQHfipz7KE=
github.com/kelindar/rate v1.0.0/go.mod h1:AjT4G+hTItNwt30lucEGZIz8y7Uk5zPho6vurIZ+1Es=
github.com/kelindar/tcp v1.0.0/go.mod h1:JB5hj1cshLU60XrLij2BBxW3JQ4hOye8vqbyvuKb52k=
github.com/logrusorgru/aurora v2.0.3+incompatible h1:tOpm7WcpBTn4fjmVfgpQq0EfczGlG91VSDkswnjF5A8=
github.com/logrusorgru/aurora v2.0.3+incompatible/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8=
github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
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/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
github.com/pion/rtp v1.7.4 h1:4dMbjb1SuynU5OpA3kz1zHK+u+eOCQjW3MAeVHf1ODA=
github.com/pion/rtp v1.7.4/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
github.com/pixelbender/go-sdp v1.1.0/go.mod h1:6IBlz9+BrUHoFTea7gcp4S54khtOhjCW/nVDLhmZBAs=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/q191201771/naza v0.19.1 h1:4KLcxT2CHztO+7miPRtBG3FFgadSQYQw1gPPPKN7rnY=
github.com/q191201771/naza v0.19.1/go.mod h1:5LeGupZZFtYP1g/S203n9vXoUNVdlRnPIfM6rExjqt0=
github.com/sqs/goreturns v0.0.0-20181028201513-538ac6014518/go.mod h1:CKI4AZ4XmGV240rTHfO0hfE83S6/a3/Q1siZJ/vXf7A=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
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-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae h1:/WDfKMnPU+m5M4xB+6x4kaepxRw6jWvR5iDRdvjHgy8=
golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

59
handshake.go
View File

@@ -1,15 +1,14 @@
package rtmp
import (
"bufio"
"bytes"
"crypto/hmac"
"crypto/sha256"
"encoding/binary"
"errors"
"fmt"
"io"
"math/rand"
"net"
)
const (
@@ -70,8 +69,8 @@ func ReadBuf(r io.Reader, length int) (buf []byte) {
return
}
func Handshake(brw *bufio.ReadWriter) error {
C0C1 := ReadBuf(brw, 1536+1)
func (nc *NetConnection) Handshake() error {
C0C1 := ReadBuf(nc.Reader, 1536+1)
if C0C1[0] != RTMP_HANDSHAKE_VERSION {
return errors.New("C0 Error")
}
@@ -85,33 +84,24 @@ func Handshake(brw *bufio.ReadWriter) error {
temp := C1[4] & 0xff
if temp == 0 {
return simple_handshake(brw, C1)
return nc.simple_handshake(C1)
}
return complex_handshake(brw, C1)
return nc.complex_handshake(C1)
}
func simple_handshake(brw *bufio.ReadWriter, C1 []byte) error {
var S0 byte
S0 = 0x03
S1 := make([]byte, 1536-4)
S2 := C1
S1_Time := uint32(0)
buf := new(bytes.Buffer)
buf.WriteByte(S0)
binary.Write(buf, binary.BigEndian, S1_Time)
buf.Write(S1)
buf.Write(S2)
brw.Write(buf.Bytes())
brw.Flush() // Don't forget to flush
ReadBuf(brw, 1536)
func (nc *NetConnection) simple_handshake(C1 []byte) error {
S1 := make([]byte, 1536+1)
S1[0] = RTMP_HANDSHAKE_VERSION
buffer := net.Buffers{S1, C1}
buffer.WriteTo(nc)
if C2 := ReadBuf(nc.Reader, 1536); bytes.Compare(C2, S1[1:]) != 0 {
return errors.New("C2 Error")
}
return nil
}
func complex_handshake(brw *bufio.ReadWriter, C1 []byte) error {
func (nc *NetConnection) complex_handshake(C1 []byte) error {
// 验证客户端,digest偏移位置和scheme由客户端定.
scheme, challenge, digest, ok, err := validateClient(C1)
if err != nil {
@@ -124,10 +114,6 @@ func complex_handshake(brw *bufio.ReadWriter, C1 []byte) error {
return errors.New("validateClient failed")
}
// s0
var S0 byte
S0 = 0x03
// s1
S1 := create_S1()
S1_Digest_Offset := scheme_Digest_Offset(S1, scheme)
@@ -163,16 +149,10 @@ func complex_handshake(brw *bufio.ReadWriter, C1 []byte) error {
return err
}
buffer := new(bytes.Buffer)
buffer.WriteByte(S0)
buffer.Write(S1)
buffer.Write(S2_Random)
buffer.Write(S2_Digest)
buffer := net.Buffers{[]byte{RTMP_HANDSHAKE_VERSION}, S1, S2_Random, S2_Digest}
buffer.WriteTo(nc)
brw.Write(buffer.Bytes())
brw.Flush()
ReadBuf(brw, 1536)
ReadBuf(nc.Reader, 1536)
return nil
}
@@ -338,8 +318,5 @@ func cerate_S2() []byte {
s2_Random[i] = byte(rand.Int() % 256)
}
buf := new(bytes.Buffer)
buf.Write(s2_Random)
return buf.Bytes()
return s2_Random
}

48
main.go
View File

@@ -1,26 +1,52 @@
package rtmp
import (
"context"
"log"
"github.com/Monibuca/engine/v3"
. "github.com/Monibuca/utils/v3"
. "github.com/Monibuca/engine/v4"
"github.com/Monibuca/engine/v4/util"
. "github.com/logrusorgru/aurora"
)
var config = struct {
type RTMPConfig struct {
Publish PublishConfig
Subscribe SubscribeConfig
ListenAddr string
ChunkSize int
}{":1935", 512}
context.Context
cancel context.CancelFunc
}
var config = &RTMPConfig{
Publish: DefaultPublishConfig,
Subscribe: DefaultSubscribeConfig,
ChunkSize: 4096,
ListenAddr: ":1935",
}
func (cfg *RTMPConfig) Update(override map[string]any) {
if cfg.cancel == nil || (override != nil && override["ListenAddr"] != nil) {
start()
}
}
func init() {
pc := engine.PluginConfig{
Name: "RTMP",
Config: &config,
InstallPlugin(config)
}
pc.Install(run)
func start() {
if config.cancel == nil {
util.Print(Green("server rtmp start at"), BrightBlue(config.ListenAddr))
} else {
config.cancel()
util.Print(Green("server rtmp restart at"), BrightBlue(config.ListenAddr))
}
config.Context, config.cancel = context.WithCancel(Ctx)
err := util.ListenTCP(config.ListenAddr, config)
if err == context.Canceled {
log.Println(err)
} else {
log.Fatal(err)
}
func run() {
Print(Green("server rtmp start at"), BrightBlue(config.ListenAddr))
log.Fatal(ListenTCP(config.ListenAddr, processRtmp))
}

312
msg.go
View File

@@ -1,12 +1,11 @@
package rtmp
import (
"bytes"
"encoding/binary"
"errors"
"log"
"sync"
"github.com/Monibuca/utils/v3"
"github.com/Monibuca/engine/v4/util"
)
const (
@@ -64,13 +63,6 @@ const (
RTMP_CSID_VIDEO = 0x05
)
var (
chunkMsgPool = &sync.Pool{
New: func() interface{} {
return new(Chunk)
},
}
)
func newChunkHeader(messageType byte) *ChunkHeader {
head := new(ChunkHeader)
@@ -107,20 +99,21 @@ type HaveStreamID interface {
}
func GetRtmpMessage(chunk *Chunk) error {
body := util.Buffer(chunk.Body)
switch chunk.MessageTypeID {
case RTMP_MSG_CHUNK_SIZE, RTMP_MSG_ABORT, RTMP_MSG_ACK, RTMP_MSG_ACK_SIZE:
if len(chunk.Body) < 4 {
return errors.New("chunk.Body < 4")
}
chunk.MsgData = Uint32Message(utils.BigEndian.Uint32(chunk.Body))
chunk.MsgData = Uint32Message(body.ReadUint32())
case RTMP_MSG_USER_CONTROL: // RTMP消息类型ID=4, 用户控制消息.客户端或服务端发送本消息通知对方用户的控制事件.
{
if len(chunk.Body) < 4 {
return errors.New("chunk.Body < 4")
}
base := UserControlMessage{
EventType: utils.BigEndian.Uint16(chunk.Body),
EventData: chunk.Body[2:],
EventType: body.ReadUint16(),
EventData: body,
}
switch base.EventType {
case RTMP_USER_STREAM_BEGIN: // 服务端向客户端发送本事件通知对方一个流开始起作用可以用于通讯.在默认情况下,服务端在成功地从客户端接收连接命令之后发送本事件,事件ID为0.事件数据是表示开始起作用的流的ID.
@@ -130,28 +123,28 @@ func GetRtmpMessage(chunk *Chunk) error {
}
if len(base.EventData) >= 4 {
//服务端在成功地从客户端接收连接命令之后发送本事件,事件ID为0.事件数据是表示开始起作用的流的ID.
m.StreamID = utils.BigEndian.Uint32(base.EventData)
m.StreamID = body.ReadUint32()
}
chunk.MsgData = m
case RTMP_USER_STREAM_EOF, RTMP_USER_STREAM_DRY, RTMP_USER_STREAM_IS_RECORDED: // 服务端向客户端发送本事件通知客户端,数据回放完成.果没有发行额外的命令,就不再发送数据.客户端丢弃从流中接收的消息.4字节的事件数据表示,回放结束的流的ID.
m := &StreamIDMessage{
UserControlMessage: base,
StreamID: utils.BigEndian.Uint32(base.EventData),
StreamID: body.ReadUint32(),
}
chunk.MsgData = m
case RTMP_USER_SET_BUFFLEN: // 客户端向服务端发送本事件,告知对方自己存储一个流的数据的缓存的长度(毫秒单位).当服务端开始处理一个流得时候发送本事件.事件数据的头四个字节表示流ID,后4个字节表示缓存长度(毫秒单位).
m := &SetBufferMessage{
StreamIDMessage: StreamIDMessage{
UserControlMessage: base,
StreamID: utils.BigEndian.Uint32(base.EventData),
StreamID: body.ReadUint32(),
},
Millisecond: utils.BigEndian.Uint32(base.EventData[4:]),
Millisecond: body.ReadUint32(),
}
chunk.MsgData = m
case RTMP_USER_PING_REQUEST: // 服务端通过本事件测试客户端是否可达.事件数据是4个字节的事件戳.代表服务调用本命令的本地时间.客户端在接收到kMsgPingRequest之后返回kMsgPingResponse事件
m := &PingRequestMessage{
UserControlMessage: base,
Timestamp: utils.BigEndian.Uint32(base.EventData),
Timestamp: body.ReadUint32(),
}
chunk.MsgData = m
case RTMP_USER_PING_RESPONSE, RTMP_USER_EMPTY: // 客户端向服务端发送本消息响应ping请求.事件数据是接kMsgPingRequest请求的时间.
@@ -165,10 +158,10 @@ func GetRtmpMessage(chunk *Chunk) error {
return errors.New("chunk.Body < 4")
}
m := &SetPeerBandwidthMessage{
AcknowledgementWindowsize: utils.BigEndian.Uint32(chunk.Body),
AcknowledgementWindowsize: body.ReadUint32(),
}
if len(chunk.Body) > 4 {
m.LimitType = chunk.Body[4]
if body.Len() > 0 {
m.LimitType = body[0]
}
chunk.MsgData = m
case RTMP_MSG_EDGE: // RTMP消息类型ID=7, 用于边缘服务与源服务器.
@@ -207,86 +200,86 @@ func GetRtmpMessage(chunk *Chunk) error {
// object类型要复杂点.
// 第一个byte是03表示object,其后跟的是N个(key+value).最后以00 00 09表示object结束
func decodeCommandAMF0(chunk *Chunk) {
amf := newAMFDecoder(chunk.Body) // rtmp_amf.go, amf 是 bytes类型, 将rtmp body(payload)放到bytes.Buffer(amf)中去.
cmd := readString(amf) // rtmp_amf.go, 将payload的bytes类型转换成string类型.
amf := AMF{chunk.Body} // rtmp_amf.go, amf 是 bytes类型, 将rtmp body(payload)放到bytes.Buffer(amf)中去.
cmd := amf.readString() // rtmp_amf.go, 将payload的bytes类型转换成string类型.
cmdMsg := CommandMessage{
cmd,
readTransactionId(amf),
uint64(amf.readNumber()),
}
switch cmd {
case "connect", "call":
chunk.MsgData = &CallMessage{
cmdMsg,
readObject(amf),
readObject(amf),
amf.readObject(),
amf.readObject(),
}
case "createStream":
amf.readNull()
chunk.MsgData = &CreateStreamMessage{
cmdMsg, readObject(amf),
cmdMsg, amf.readObject(),
}
case "play":
amf.readNull()
chunk.MsgData = &PlayMessage{
cmdMsg,
readString(amf),
readNumber(amf),
readNumber(amf),
readBool(amf),
amf.readString(),
uint64(amf.readNumber()),
uint64(amf.readNumber()),
amf.readBool(),
}
case "play2":
amf.readNull()
chunk.MsgData = &Play2Message{
cmdMsg,
readNumber(amf),
readString(amf),
readString(amf),
readNumber(amf),
readString(amf),
uint64(amf.readNumber()),
amf.readString(),
amf.readString(),
uint64(amf.readNumber()),
amf.readString(),
}
case "publish":
amf.readNull()
chunk.MsgData = &PublishMessage{
cmdMsg,
readString(amf),
readString(amf),
amf.readString(),
amf.readString(),
}
case "pause":
amf.readNull()
chunk.MsgData = &PauseMessage{
cmdMsg,
readBool(amf),
readNumber(amf),
amf.readBool(),
uint64(amf.readNumber()),
}
case "seek":
amf.readNull()
chunk.MsgData = &SeekMessage{
cmdMsg,
readNumber(amf),
uint64(amf.readNumber()),
}
case "deleteStream", "closeStream":
amf.readNull()
chunk.MsgData = &CURDStreamMessage{
cmdMsg,
uint32(readNumber(amf)),
uint32(amf.readNumber()),
}
case "releaseStream":
amf.readNull()
chunk.MsgData = &ReleaseStreamMessage{
cmdMsg,
readString(amf),
amf.readString(),
}
case "receiveAudio", "receiveVideo":
amf.readNull()
chunk.MsgData = &ReceiveAVMessage{
cmdMsg,
readBool(amf),
amf.readBool(),
}
case "_result", "_error", "onStatus":
chunk.MsgData = &ResponseMessage{
cmdMsg,
readObject(amf),
readObject(amf), "",
amf.readObject(),
amf.readObject(), "",
}
case "FCPublish", "FCUnpublish":
default:
@@ -299,28 +292,6 @@ func decodeCommandAMF3(chunk *Chunk) {
decodeCommandAMF0(chunk)
}
func readTransactionId(amf *AMF) uint64 {
v, _ := amf.readNumber()
return uint64(v)
}
func readString(amf *AMF) string {
v, _ := amf.readString()
return v
}
func readNumber(amf *AMF) uint64 {
v, _ := amf.readNumber()
return uint64(v)
}
func readBool(amf *AMF) bool {
v, _ := amf.readBool()
return v
}
func readObject(amf *AMF) AMFObjects {
v, _ := amf.readObject()
return v
}
/* Command Message */
type CommandMessage struct {
CommandName string // 命令名. 字符串. 命令名.设置为"connect"
@@ -333,12 +304,13 @@ type Commander interface {
func (cmd *CommandMessage) GetCommand() *CommandMessage {
return cmd
}
func (msg *CommandMessage) Encode() (b []byte) {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeNull()
return amf.Bytes()
return amf.Buffer
}
// Protocol control message 1.
@@ -349,7 +321,7 @@ type Uint32Message uint32
func (msg Uint32Message) Encode() (b []byte) {
b = make([]byte, 4)
utils.BigEndian.PutUint32(b, uint32(msg))
binary.BigEndian.PutUint32(b, uint32(msg))
return b
}
@@ -376,7 +348,7 @@ type SetPeerBandwidthMessage struct {
func (msg *SetPeerBandwidthMessage) Encode() (b []byte) {
b = make([]byte, 5)
utils.BigEndian.PutUint32(b, msg.AcknowledgementWindowsize)
binary.BigEndian.PutUint32(b, msg.AcknowledgementWindowsize)
b[4] = msg.LimitType
return
}
@@ -404,30 +376,27 @@ type MetadataMessage struct {
// The called RPC name is passed as a parameter to the call command.
type CallMessage struct {
CommandMessage
Object interface{} `json:",omitempty"`
Optional interface{} `json:",omitempty"`
Object AMFObject `json:",omitempty"`
Optional AMFObject `json:",omitempty"`
}
func (msg *CallMessage) Encode() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
if msg.Object != nil {
amf.encodeObject(msg.Object.(AMFObjects))
}
if msg.Optional != nil {
amf.encodeObject(msg.Optional.(AMFObjects))
}
return amf.Bytes()
amf.writeObject(msg.Object)
amf.writeObject(msg.Optional)
return amf.Buffer
}
func (msg *CallMessage) Encode3() []byte {
buf := new(bytes.Buffer)
buf.WriteByte(0)
buf.Write(msg.Encode())
return buf.Bytes()
var amf AMF
amf.WriteUint8(0)
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeObject(msg.Object)
amf.writeObject(msg.Optional)
return amf.Buffer
}
// Create Stream Message.
@@ -436,18 +405,15 @@ func (msg *CallMessage) Encode3() []byte {
type CreateStreamMessage struct {
CommandMessage
Object interface{}
Object AMFObject
}
func (msg *CreateStreamMessage) Encode() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
if msg.Object != nil {
amf.encodeObject(msg.Object.(AMFObjects))
}
return amf.Bytes()
amf.writeObject(msg.Object)
return amf.Buffer
}
/*
@@ -493,7 +459,7 @@ type PlayMessage struct {
// Reset -> 可选的布尔值或者数字定义了是否对以前的播放列表进行 flush
func (msg *PlayMessage) Encode() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeNull()
@@ -507,7 +473,7 @@ func (msg *PlayMessage) Encode() []byte {
}
amf.writeBool(msg.Rest)
return amf.Bytes()
return amf.Buffer
}
/*
@@ -615,23 +581,17 @@ func (msg *PauseMessage) Encode0() {
//
type ResponseConnectMessage struct {
CommandMessage
Properties interface{} `json:",omitempty"`
Infomation interface{} `json:",omitempty"`
Properties AMFObject `json:",omitempty"`
Infomation AMFObject `json:",omitempty"`
}
func (msg *ResponseConnectMessage) Encode() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
if msg.Properties != nil {
amf.encodeObject(msg.Properties.(AMFObjects))
}
if msg.Infomation != nil {
amf.encodeObject(msg.Infomation.(AMFObjects))
}
return amf.Bytes()
amf.writeObject(msg.Properties)
amf.writeObject(msg.Infomation)
return amf.Buffer
}
/*
@@ -642,23 +602,17 @@ func (msg *ResponseConnectMessage) Encode3() {
//
type ResponseCallMessage struct {
CommandMessage
Object interface{}
Response interface{}
Object AMFObject
Response AMFObject
}
func (msg *ResponseCallMessage) Encode0() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
if msg.Object != nil {
amf.encodeObject(msg.Object.(AMFObjects))
}
if msg.Response != nil {
amf.encodeObject(msg.Response.(AMFObjects))
}
return amf.Bytes()
amf.writeObject(msg.Object)
amf.writeObject(msg.Response)
return amf.Buffer
}
//
@@ -671,12 +625,12 @@ type ResponseCreateStreamMessage struct {
}
func (msg *ResponseCreateStreamMessage) Encode() []byte {
amf := newAMFEncoder() // rtmp_amf.go
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeNull()
amf.writeNumber(float64(msg.StreamId))
return amf.Bytes()
return amf.Buffer
}
/*
@@ -684,18 +638,11 @@ func (msg *ResponseCreateStreamMessage) Encode3() {
}*/
func (msg *ResponseCreateStreamMessage) Decode0(chunk *Chunk) {
amf := newAMFDecoder(chunk.Body)
if obj, err := amf.decodeObject(); err == nil {
msg.CommandName = obj.(string)
}
if obj, err := amf.decodeObject(); err == nil {
msg.TransactionId = uint64(obj.(float64))
}
amf := AMF{chunk.Body}
msg.CommandName = amf.decodeObject().(string)
msg.TransactionId = uint64(amf.decodeObject().(float64))
amf.decodeObject()
if obj, err := amf.decodeObject(); err == nil {
msg.StreamId = uint32(obj.(float64))
}
msg.StreamId = uint32(amf.decodeObject().(float64))
}
func (msg *ResponseCreateStreamMessage) Decode3(chunk *Chunk) {
chunk.Body = chunk.Body[1:]
@@ -707,7 +654,7 @@ func (msg *ResponseCreateStreamMessage) Decode3(chunk *Chunk) {
//
type ResponsePlayMessage struct {
CommandMessage
Object interface{} `json:",omitempty"`
Object AMFObject `json:",omitempty"`
Description string
StreamID uint32
}
@@ -716,15 +663,13 @@ func (msg *ResponsePlayMessage) GetStreamID() uint32 {
return msg.StreamID
}
func (msg *ResponsePlayMessage) Encode() []byte {
amf := newAMFEncoder() // rtmp_amf.go
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeNull()
if msg.Object != nil {
amf.encodeObject(msg.Object.(AMFObjects))
}
amf.writeObject(msg.Object)
amf.writeString(msg.Description)
return amf.Bytes()
return amf.Buffer
}
/*
@@ -732,20 +677,10 @@ func (msg *ResponsePlayMessage) Encode3() {
}*/
func (msg *ResponsePlayMessage) Decode0(chunk *Chunk) {
amf := newAMFDecoder(chunk.Body)
if obj, err := amf.decodeObject(); err == nil {
msg.CommandName = obj.(string)
}
if obj, err := amf.decodeObject(); err == nil {
msg.TransactionId = uint64(obj.(float64))
}
obj, err := amf.decodeObject()
if err == nil && obj != nil {
msg.Object = obj
} else if obj, err := amf.decodeObject(); err == nil {
msg.Object = obj
}
amf := AMF{chunk.Body}
msg.CommandName = amf.decodeObject().(string)
msg.TransactionId = uint64(amf.decodeObject().(float64))
msg.Object = amf.decodeObject().(AMFObject)
}
func (msg *ResponsePlayMessage) Decode3(chunk *Chunk) {
chunk.Body = chunk.Body[1:]
@@ -757,8 +692,8 @@ func (msg *ResponsePlayMessage) Decode3(chunk *Chunk) {
//
type ResponsePublishMessage struct {
CommandMessage
Properties interface{} `json:",omitempty"`
Infomation interface{} `json:",omitempty"`
Properties AMFObject `json:",omitempty"`
Infomation AMFObject `json:",omitempty"`
StreamID uint32
}
@@ -772,19 +707,13 @@ func (msg *ResponsePublishMessage) GetStreamID() uint32 {
// 信息 -> level, code, description
func (msg *ResponsePublishMessage) Encode() []byte {
amf := newAMFEncoder()
var amf AMF
amf.writeString(msg.CommandName)
amf.writeNumber(float64(msg.TransactionId))
amf.writeNull()
if msg.Properties != nil {
amf.encodeObject(msg.Properties.(AMFObjects))
}
if msg.Infomation != nil {
amf.encodeObject(msg.Infomation.(AMFObjects))
}
return amf.Bytes()
amf.writeObject(msg.Properties)
amf.writeObject(msg.Infomation)
return amf.Buffer
}
/*
@@ -828,8 +757,8 @@ func (msg *ResponsePauseMessage) Encode0() {
//
type ResponseMessage struct {
CommandMessage
Properties interface{} `json:",omitempty"`
Infomation interface{} `json:",omitempty"`
Properties any `json:",omitempty"`
Infomation any `json:",omitempty"`
Description string
}
@@ -840,13 +769,9 @@ func (msg *ResponseMessage) Encode0() {
//}
func (msg *ResponseMessage) Decode0(chunk *Chunk) {
amf := newAMFDecoder(chunk.Body)
if obj, err := amf.decodeObject(); err == nil {
msg.CommandName = obj.(string)
}
if obj, err := amf.decodeObject(); err == nil {
msg.TransactionId = uint64(obj.(float64))
}
amf := AMF{chunk.Body}
msg.CommandName = amf.decodeObject().(string)
msg.TransactionId = uint64(amf.decodeObject().(float64))
}
// User Control Message 4.
@@ -866,11 +791,11 @@ type StreamIDMessage struct {
}
func (msg *StreamIDMessage) Encode() (b []byte) {
b = make([]byte, 6)
utils.BigEndian.PutUint16(b, msg.EventType)
utils.BigEndian.PutUint32(b[2:], msg.StreamID)
msg.EventData = b[2:]
return
buf := util.Buffer(make([]byte, 0, 6))
buf.WriteUint16(msg.EventType)
buf.WriteUint32(msg.StreamID)
msg.EventData = buf[2:]
return buf
}
// SetBuffer Length (=3)
@@ -884,12 +809,12 @@ type SetBufferMessage struct {
}
func (msg *SetBufferMessage) Encode() []byte {
b := make([]byte, 10)
utils.BigEndian.PutUint16(b, msg.EventType)
utils.BigEndian.PutUint32(b[2:], msg.StreamID)
utils.BigEndian.PutUint32(b[6:], msg.Millisecond)
msg.EventData = b[2:]
return b
buf := util.Buffer(make([]byte, 0, 10))
buf.WriteUint16(msg.EventType)
buf.WriteUint32(msg.StreamID)
buf.WriteUint32(msg.Millisecond)
msg.EventData = buf[2:]
return buf
}
// PingRequest (=6)
@@ -902,18 +827,15 @@ type PingRequestMessage struct {
}
func (msg *PingRequestMessage) Encode() (b []byte) {
b = make([]byte, 6)
utils.BigEndian.PutUint16(b, msg.EventType)
utils.BigEndian.PutUint32(b[2:], msg.Timestamp)
msg.EventData = b[2:]
return
buf := util.Buffer(make([]byte, 0, 6))
buf.WriteUint16(msg.EventType)
buf.WriteUint32(msg.Timestamp)
msg.EventData = buf[2:]
return buf
}
func (msg *UserControlMessage) Encode() []byte {
b := make([]byte, 2)
utils.BigEndian.PutUint16(b, msg.EventType)
msg.EventData = b[2:]
return b
return util.PutBE(make([]byte, 2), msg.EventType)
}
type AVPack struct {

476
netConnection.go
View File

@@ -2,12 +2,13 @@ package rtmp
import (
"bufio"
"encoding/binary"
"errors"
"io"
"log"
"net"
"github.com/Monibuca/engine/v3"
"github.com/Monibuca/utils/v3"
"github.com/Monibuca/engine/v4"
"github.com/Monibuca/engine/v4/util"
)
const (
@@ -43,8 +44,8 @@ const (
SEND_FULL_VDIEO_MESSAGE = "Send Full Video Message"
)
func newConnectResponseMessageData(objectEncoding float64) (amfobj AMFObjects) {
amfobj = newAMFObjects()
func newConnectResponseMessageData(objectEncoding float64) (amfobj AMFObject) {
amfobj = make(AMFObject)
amfobj["fmsVer"] = "monibuca/" + engine.Version
amfobj["capabilities"] = 31
amfobj["mode"] = 1
@@ -56,8 +57,8 @@ func newConnectResponseMessageData(objectEncoding float64) (amfobj AMFObjects) {
return
}
func newPublishResponseMessageData(streamid uint32, code, level string) (amfobj AMFObjects) {
amfobj = newAMFObjects()
func newPublishResponseMessageData(streamid uint32, code, level string) (amfobj AMFObject) {
amfobj = make(AMFObject)
amfobj["code"] = code
amfobj["level"] = level
amfobj["streamid"] = streamid
@@ -65,8 +66,8 @@ func newPublishResponseMessageData(streamid uint32, code, level string) (amfobj
return
}
func newPlayResponseMessageData(streamid uint32, code, level string) (amfobj AMFObjects) {
amfobj = newAMFObjects()
func newPlayResponseMessageData(streamid uint32, code, level string) (amfobj AMFObject) {
amfobj = make(AMFObject)
amfobj["code"] = code
amfobj["level"] = level
amfobj["streamid"] = streamid
@@ -75,7 +76,10 @@ func newPlayResponseMessageData(streamid uint32, code, level string) (amfobj AMF
}
type NetConnection struct {
*bufio.ReadWriter
engine.Publisher
subscribers map[uint32]*engine.Subscriber
*bufio.Reader
*net.TCPConn
bandwidth uint32
readSeqNum uint32 // 当前读的字节
writeSeqNum uint32 // 当前写的字节
@@ -83,98 +87,45 @@ type NetConnection struct {
totalRead uint32 // 总共读了多少字节
writeChunkSize int
readChunkSize int
incompleteRtmpBody map[uint32][]byte // 完整的RtmpBody,在网络上是被分成一块一块的,需要将其组装起来
nextStreamID func() uint32 // 下一个流ID
incompleteRtmpBody map[uint32]util.Buffer // 完整的RtmpBody,在网络上是被分成一块一块的,需要将其组装起来
streamID uint32 // 流ID
rtmpHeader map[uint32]*ChunkHeader // RtmpHeader
objectEncoding float64
appName string
tmpBuf []byte //用来接收小数据，复用内存
}
func (conn *NetConnection) OnConnect() (err error) {
var msg *Chunk
if msg, err = conn.RecvMessage(); err == nil {
defer chunkMsgPool.Put(msg)
if connect, ok := msg.MsgData.(*CallMessage); ok {
if connect.CommandName == "connect" {
app := DecodeAMFObject(connect.Object, "app") // 客户端要连接到的服务应用名
objectEncoding := DecodeAMFObject(connect.Object, "objectEncoding") // AMF编码方法
if objectEncoding != nil {
conn.objectEncoding = objectEncoding.(float64)
}
conn.appName = app.(string)
log.Printf("app:%v,objectEncoding:%v", app, objectEncoding)
err = conn.SendMessage(SEND_ACK_WINDOW_SIZE_MESSAGE, uint32(512<<10))
err = conn.SendMessage(SEND_SET_PEER_BANDWIDTH_MESSAGE, uint32(512<<10))
err = conn.SendMessage(SEND_STREAM_BEGIN_MESSAGE, nil)
err = conn.SendMessage(SEND_CONNECT_RESPONSE_MESSAGE, conn.objectEncoding)
return
func (conn *NetConnection) Close() {
conn.Publisher.Stream.UnPublish()
conn.TCPConn.Close()
for _, s := range conn.subscribers {
s.Close()
}
}
} else if msg != nil {
utils.Printf("recv MessageTypeID:%d error:%v", msg.MessageTypeID, err)
func (conn *NetConnection) ReadFull(buf []byte) (n int, err error) {
return io.ReadFull(conn.Reader, buf)
}
return
func (conn *NetConnection) SendStreamID0(eventType uint16) (err error) {
return conn.SendMessage(RTMP_MSG_USER_CONTROL, &StreamIDMessage{UserControlMessage{EventType: eventType}, 0})
}
func (conn *NetConnection) SendMessage(message string, args interface{}) error {
func (conn *NetConnection) SendStreamID(eventType uint16) (err error) {
return conn.SendMessage(RTMP_MSG_USER_CONTROL, &StreamIDMessage{UserControlMessage{EventType: eventType}, conn.streamID})
}
func (conn *NetConnection) SendUserControl(eventType uint16) error {
return conn.SendMessage(RTMP_MSG_USER_CONTROL, &UserControlMessage{EventType: eventType})
}
func (conn *NetConnection) SendCommand(message string, args any) error {
switch message {
case SEND_CHUNK_SIZE_MESSAGE:
size, ok := args.(uint32)
if !ok {
return errors.New(SEND_CHUNK_SIZE_MESSAGE + ", The parameter only one(size uint32)!")
}
return conn.writeMessage(RTMP_MSG_CHUNK_SIZE, Uint32Message(size))
case SEND_ACK_MESSAGE:
num, ok := args.(uint32)
if !ok {
return errors.New(SEND_ACK_MESSAGE + ", The parameter only one(number uint32)!")
}
return conn.writeMessage(RTMP_MSG_ACK, Uint32Message(num))
case SEND_ACK_WINDOW_SIZE_MESSAGE:
size, ok := args.(uint32)
if !ok {
return errors.New(SEND_ACK_WINDOW_SIZE_MESSAGE + ", The parameter only one(size uint32)!")
}
return conn.writeMessage(RTMP_MSG_ACK_SIZE, Uint32Message(size))
case SEND_SET_PEER_BANDWIDTH_MESSAGE:
size, ok := args.(uint32)
if !ok {
return errors.New(SEND_SET_PEER_BANDWIDTH_MESSAGE + ", The parameter only one(size uint32)!")
}
return conn.writeMessage(RTMP_MSG_BANDWIDTH, &SetPeerBandwidthMessage{
AcknowledgementWindowsize: size,
LimitType: byte(2),
})
case SEND_STREAM_BEGIN_MESSAGE:
if args != nil {
return errors.New(SEND_STREAM_BEGIN_MESSAGE + ", The parameter is nil")
}
return conn.writeMessage(RTMP_MSG_USER_CONTROL, &StreamIDMessage{UserControlMessage{EventType: RTMP_USER_STREAM_BEGIN}, conn.streamID})
case SEND_STREAM_IS_RECORDED_MESSAGE:
if args != nil {
return errors.New(SEND_STREAM_IS_RECORDED_MESSAGE + ", The parameter is nil")
}
return conn.writeMessage(RTMP_MSG_USER_CONTROL, &StreamIDMessage{UserControlMessage{EventType: RTMP_USER_STREAM_IS_RECORDED}, conn.streamID})
case SEND_SET_BUFFER_LENGTH_MESSAGE:
if args != nil {
return errors.New(SEND_SET_BUFFER_LENGTH_MESSAGE + ", The parameter is nil")
}
m := new(SetBufferMessage)
m.EventType = RTMP_USER_SET_BUFFLEN
m.Millisecond = 100
m.StreamID = conn.streamID
return conn.writeMessage(RTMP_MSG_USER_CONTROL, m)
case SEND_PING_REQUEST_MESSAGE:
if args != nil {
return errors.New(SEND_PING_REQUEST_MESSAGE + ", The parameter is nil")
}
return conn.writeMessage(RTMP_MSG_USER_CONTROL, &PingRequestMessage{UserControlMessage{EventType: RTMP_USER_PING_REQUEST}, 0})
case SEND_PING_RESPONSE_MESSAGE:
if args != nil {
return errors.New(SEND_PING_RESPONSE_MESSAGE + ", The parameter is nil")
}
return conn.writeMessage(RTMP_MSG_USER_CONTROL, &UserControlMessage{EventType: RTMP_USER_PING_RESPONSE})
// case SEND_SET_BUFFER_LENGTH_MESSAGE:
// if args != nil {
// return errors.New(SEND_SET_BUFFER_LENGTH_MESSAGE + ", The parameter is nil")
// }
// m := new(SetBufferMessage)
// m.EventType = RTMP_USER_SET_BUFFLEN
// m.Millisecond = 100
// m.StreamID = conn.streamID
// return conn.writeMessage(RTMP_MSG_USER_CONTROL, m)
case SEND_CREATE_STREAM_MESSAGE:
if args != nil {
return errors.New(SEND_CREATE_STREAM_MESSAGE + ", The parameter is nil")
@@ -183,7 +134,7 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
m := &CreateStreamMessage{}
m.CommandName = "createStream"
m.TransactionId = 1
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_CREATE_STREAM_RESPONSE_MESSAGE:
tid, ok := args.(uint64)
if !ok {
@@ -193,7 +144,7 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
m.CommandName = Response_Result
m.TransactionId = tid
m.StreamId = conn.streamID
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_PLAY_MESSAGE:
data, ok := args.(map[interface{}]interface{})
if !ok {
@@ -213,14 +164,14 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
m.Rest = v.(bool)
}
}
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_PLAY_RESPONSE_MESSAGE:
data, ok := args.(AMFObjects)
data, ok := args.(AMFObject)
if !ok {
errors.New(SEND_PLAY_RESPONSE_MESSAGE + ", The parameter is AMFObjects(map[string]interface{})")
}
obj := newAMFObjects()
obj := make(AMFObject)
var streamID uint32
for i, v := range data {
@@ -241,14 +192,14 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
m.TransactionId = 0
m.Object = obj
m.StreamID = streamID
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_CONNECT_RESPONSE_MESSAGE:
//if !ok {
// errors.New(SEND_CONNECT_RESPONSE_MESSAGE + ", The parameter is AMFObjects(map[string]interface{})")
//}
pro := newAMFObjects()
info := newAMFObjects()
pro := make(AMFObject)
info := make(AMFObject)
//for i, v := range data {
// switch i {
@@ -264,21 +215,21 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
info["level"] = Level_Status
info["code"] = NetConnection_Connect_Success
info["objectEncoding"] = args.(float64)
info["objectEncoding"] = conn.objectEncoding
m := new(ResponseConnectMessage)
m.CommandName = Response_Result
m.TransactionId = 1
m.Properties = pro
m.Infomation = info
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_CONNECT_MESSAGE:
data, ok := args.(AMFObjects)
data, ok := args.(AMFObject)
if !ok {
errors.New(SEND_CONNECT_MESSAGE + ", The parameter is AMFObjects(map[string]interface{})")
}
obj := newAMFObjects()
info := newAMFObjects()
obj := make(AMFObject)
info := make(AMFObject)
for i, v := range data {
switch i {
@@ -292,120 +243,67 @@ func (conn *NetConnection) SendMessage(message string, args interface{}) error {
m.TransactionId = 1
m.Object = obj
m.Optional = info
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_PUBLISH_RESPONSE_MESSAGE, SEND_PUBLISH_START_MESSAGE:
data, ok := args.(AMFObjects)
if !ok {
errors.New(SEND_CONNECT_MESSAGE + "or" + SEND_PUBLISH_START_MESSAGE + ", The parameter is AMFObjects(map[string]interface{})")
}
info := newAMFObjects()
var streamID uint32
for i, v := range data {
switch i {
case "code", "level":
info[i] = v
case "streamid":
if t, ok := v.(uint32); ok {
streamID = t
}
}
}
info := args.(AMFObject)
info["clientid"] = 1
m := new(ResponsePublishMessage)
m.CommandName = Response_OnStatus
m.TransactionId = 0
m.Infomation = info
m.StreamID = streamID
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
m.StreamID = info["streamid"].(uint32)
delete(info, "streamid")
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_UNPUBLISH_RESPONSE_MESSAGE:
data, ok := args.(AMFObjects)
data, ok := args.(AMFObject)
if !ok {
errors.New(SEND_UNPUBLISH_RESPONSE_MESSAGE + ", The parameter is AMFObjects(map[string]interface{})")
}
m := new(CommandMessage)
m.TransactionId = data["tid"].(uint64)
m.CommandName = "releaseStream" + data["level"].(string)
return conn.writeMessage(RTMP_MSG_AMF0_COMMAND, m)
case SEND_FULL_AUDIO_MESSAGE:
audio, ok := args.(*AVPack)
if !ok {
errors.New(message + ", The parameter is AVPacket")
return conn.SendMessage(RTMP_MSG_AMF0_COMMAND, m)
}
return conn.sendAVMessage(audio.Timestamp, audio.Payload, true, true)
case SEND_AUDIO_MESSAGE:
audio, ok := args.(*AVPack)
if !ok {
errors.New(message + ", The parameter is AVPacket")
}
return conn.sendAVMessage(audio.Timestamp, audio.Payload, true, false)
case SEND_FULL_VDIEO_MESSAGE:
video, ok := args.(*AVPack)
if !ok {
errors.New(message + ", The parameter is AVPacket")
}
return conn.sendAVMessage(video.Timestamp, video.Payload, false, true)
case SEND_VIDEO_MESSAGE:
video, ok := args.(*AVPack)
if !ok {
errors.New(message + ", The parameter is AVPacket")
}
return conn.sendAVMessage(video.Timestamp, video.Payload, false, false)
}
return errors.New("send message no exist")
}
// 当发送音视频数据的时候,当块类型为12的时候,Chunk Message Header有一个字段TimeStamp,指明一个时间
// 当块类型为4,8的时候,Chunk Message Header有一个字段TimeStamp Delta,记录与上一个Chunk的时间差值
// 当块类型为0的时候,Chunk Message Header没有时间字段,与上一个Chunk时间值相同
func (conn *NetConnection) sendAVMessage(ts uint32, payload []byte, isAudio bool, isFirst bool) error {
func (conn *NetConnection) sendAVMessage(ts uint32, payload net.Buffers, isAudio bool, isFirst bool) (err error) {
if conn.writeSeqNum > conn.bandwidth {
conn.totalWrite += conn.writeSeqNum
conn.writeSeqNum = 0
conn.SendMessage(SEND_ACK_MESSAGE, conn.totalWrite)
conn.SendMessage(SEND_PING_REQUEST_MESSAGE, nil)
conn.SendMessage(RTMP_MSG_ACK, Uint32Message(conn.totalWrite))
conn.SendStreamID0(RTMP_USER_PING_REQUEST)
}
var err error
var need []byte
var head *ChunkHeader
if isAudio {
head = newRtmpHeader(RTMP_CSID_AUDIO, ts, uint32(len(payload)), RTMP_MSG_AUDIO, conn.streamID, 0)
head = newRtmpHeader(RTMP_CSID_AUDIO, ts, uint32(util.SizeOfBuffers(payload)), RTMP_MSG_AUDIO, conn.streamID, 0)
} else {
head = newRtmpHeader(RTMP_CSID_VIDEO, ts, uint32(len(payload)), RTMP_MSG_VIDEO, conn.streamID, 0)
head = newRtmpHeader(RTMP_CSID_VIDEO, ts, uint32(util.SizeOfBuffers(payload)), RTMP_MSG_VIDEO, conn.streamID, 0)
}
// 第一次是发送关键帧,需要完整的消息头(Chunk Basic Header(1) + Chunk Message Header(11) + Extended Timestamp(4)(可能会要包括))
// 后面开始,就是直接发送音视频数据,那么直接发送,不需要完整的块(Chunk Basic Header(1) + Chunk Message Header(7))
// 当Chunk Type为0时(即Chunk12),
var chunk1 net.Buffers
if isFirst {
need, err = conn.encodeChunk12(head, payload)
chunk1 = append(chunk1, conn.encodeChunk12(head))
} else {
need, err = conn.encodeChunk8(head, payload)
chunk1 = append(chunk1, conn.encodeChunk8(head))
}
if err != nil {
return err
}
if err = conn.Flush(); err != nil {
return err
}
chunks := util.SplitBuffers(payload, conn.writeChunkSize)
chunk1 = append(chunk1, chunks[0]...)
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk1))
_, err = chunk1.WriteTo(conn)
// 如果在音视频数据太大,一次发送不完,那么这里进行分割(data + Chunk Basic Header(1))
for len(need) > 0 {
if need, err = conn.encodeChunk1(head, need); err != nil {
return err
}
if err = conn.Flush(); err != nil {
return err
}
for _, chunk := range chunks[1:] {
chunk1 = net.Buffers{conn.encodeChunk1(head)}
chunk1 = append(chunk1, chunk...)
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk1))
_, err = chunk1.WriteTo(conn)
}
return nil
@@ -429,56 +327,49 @@ func (conn *NetConnection) readChunk() (msg *Chunk, err error) {
h, ok := conn.rtmpHeader[ChunkStreamID]
if !ok {
h = new(ChunkHeader)
h.ChunkStreamID = ChunkStreamID
h.ChunkType = ChunkType
h = &ChunkHeader{
ChunkBasicHeader: ChunkBasicHeader{
ChunkStreamID,
ChunkType,
},
}
conn.rtmpHeader[ChunkStreamID] = h
}
currentBody, ok := conn.incompleteRtmpBody[ChunkStreamID]
if ChunkType != 3 && ok {
if ChunkType != 3 && ok && currentBody.Len() > 0 {
// 如果块类型不为3,那么这个rtmp的body应该为空.
return nil, errors.New("incompleteRtmpBody error")
}
chunkHead, err := conn.readChunkType(h, ChunkType)
if err != nil {
if err = conn.readChunkType(h, ChunkType); err != nil {
return nil, errors.New("get chunk type error :" + err.Error())
}
msgLen := int(chunkHead.MessageLength)
if !ok {
currentBody = make([]byte, 0, msgLen)
msgLen := int(h.MessageLength)
if !ok || currentBody.Len() == 0 {
currentBody = util.Buffer(make([]byte, 0, msgLen))
conn.incompleteRtmpBody[ChunkStreamID] = currentBody
}
markRead := len(currentBody)
needRead := conn.readChunkSize
unRead := msgLen - markRead
if unRead < needRead {
if unRead := msgLen - currentBody.Len(); unRead < needRead {
needRead = unRead
}
if n, err := io.ReadFull(conn, currentBody[markRead:needRead+markRead]); err != nil {
if n, err := conn.ReadFull(currentBody.Malloc(needRead)); err != nil {
util.Println(err)
return nil, err
} else {
markRead += n
conn.readSeqNum += uint32(n)
}
currentBody = currentBody[:markRead]
conn.incompleteRtmpBody[ChunkStreamID] = currentBody
// 如果读完了一个完整的块,那么就返回这个消息,没读完继续递归读块.
if markRead == msgLen {
msg := chunkMsgPool.Get().(*Chunk)
msg.MsgData = nil
msg.Body = currentBody
msg.ChunkHeader = chunkHead.Clone()
err = GetRtmpMessage(msg)
delete(conn.incompleteRtmpBody, ChunkStreamID)
return msg, err
if currentBody.Len() == msgLen {
msg = &Chunk{
ChunkHeader: *h,
Body: currentBody.ReadN(msgLen),
}
return conn.readChunk()
err = GetRtmpMessage(msg)
}
conn.incompleteRtmpBody[ChunkStreamID] = currentBody
return
}
func (conn *NetConnection) readChunkStreamID(csid uint32) (chunkStreamID uint32, err error) {
@@ -513,108 +404,99 @@ func (conn *NetConnection) readChunkStreamID(csid uint32) (chunkStreamID uint32,
return chunkStreamID, nil
}
func (conn *NetConnection) readChunkType(h *ChunkHeader, chunkType byte) (head *ChunkHeader, err error) {
func (conn *NetConnection) readChunkType(h *ChunkHeader, chunkType byte) (err error) {
b := conn.tmpBuf[:3]
switch chunkType {
case 0:
{
// Timestamp 3 bytes
b := utils.GetSlice(3)
if _, err := io.ReadFull(conn, b); err != nil {
return nil, err
if _, err := conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 3
h.Timestamp = utils.BigEndian.Uint24(b) //type = 0的时间戳为绝对时间,其他的都为相对时间
util.GetBE(b, &h.Timestamp) //type = 0的时间戳为绝对时间,其他的都为相对时间
// Message Length 3 bytes
if _, err = io.ReadFull(conn, b); err != nil { // 读取Message Length,这里的长度指的是一条信令或者一帧视频数据或音频数据的长度,而不是Chunk data的长度.
return nil, err
if _, err = conn.ReadFull(b); err != nil { // 读取Message Length,这里的长度指的是一条信令或者一帧视频数据或音频数据的长度,而不是Chunk data的长度.
return err
}
conn.readSeqNum += 3
h.MessageLength = utils.BigEndian.Uint24(b)
utils.RecycleSlice(b)
util.GetBE(b, &h.MessageLength)
// Message Type ID 1 bytes
v, err := conn.ReadByte() // 读取Message Type ID
if err != nil {
return nil, err
return err
}
conn.readSeqNum++
h.MessageTypeID = v
// Message Stream ID 4bytes
bb := utils.GetSlice(4)
if _, err = io.ReadFull(conn, bb); err != nil { // 读取Message Stream ID
return nil, err
b = conn.tmpBuf
if _, err = conn.ReadFull(b); err != nil { // 读取Message Stream ID
return err
}
conn.readSeqNum += 4
h.MessageStreamID = utils.LittleEndian.Uint32(bb)
h.MessageStreamID = binary.LittleEndian.Uint32(b)
// ExtendTimestamp 4 bytes
if h.Timestamp == 0xffffff { // 对于type 0的chunk,绝对时间戳在这里表示,如果时间戳值大于等于0xffffff(16777215),该值必须是0xffffff,且时间戳扩展字段必须发送,其他情况没有要求
if _, err = io.ReadFull(conn, bb); err != nil {
return nil, err
if _, err = conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 4
h.ExtendTimestamp = utils.BigEndian.Uint32(bb)
util.GetBE(b, &h.ExtendTimestamp)
}
utils.RecycleSlice(bb)
}
case 1:
{
// Timestamp 3 bytes
b := utils.GetSlice(3)
if _, err = io.ReadFull(conn, b); err != nil {
return nil, err
if _, err = conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 3
h.ChunkType = chunkType
h.Timestamp = utils.BigEndian.Uint24(b)
util.GetBE(b, &h.Timestamp)
// Message Length 3 bytes
if _, err = io.ReadFull(conn, b); err != nil {
return nil, err
if _, err = conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 3
h.MessageLength = utils.BigEndian.Uint24(b)
utils.RecycleSlice(b)
util.GetBE(b, &h.MessageLength)
// Message Type ID 1 bytes
v, err := conn.ReadByte()
if err != nil {
return nil, err
return err
}
conn.readSeqNum++
h.MessageTypeID = v
// ExtendTimestamp 4 bytes
if h.Timestamp == 0xffffff {
bb := utils.GetSlice(4)
if _, err := io.ReadFull(conn, bb); err != nil {
return nil, err
b = conn.tmpBuf
if _, err := conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 4
h.ExtendTimestamp = utils.BigEndian.Uint32(bb)
utils.RecycleSlice(bb)
util.GetBE(b, &h.ExtendTimestamp)
}
}
case 2:
{
// Timestamp 3 bytes
b := utils.GetSlice(3)
if _, err = io.ReadFull(conn, b); err != nil {
return nil, err
if _, err = conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 3
h.ChunkType = chunkType
h.Timestamp = utils.BigEndian.Uint24(b)
utils.RecycleSlice(b)
util.GetBE(b, &h.Timestamp)
// ExtendTimestamp 4 bytes
if h.Timestamp == 0xffffff {
bb := utils.GetSlice(4)
if _, err := io.ReadFull(conn, bb); err != nil {
return nil, err
b = conn.tmpBuf
if _, err := conn.ReadFull(b); err != nil {
return err
}
conn.readSeqNum += 4
h.ExtendTimestamp = utils.BigEndian.Uint32(bb)
utils.RecycleSlice(bb)
util.GetBE(b, &h.ExtendTimestamp)
}
}
case 3:
@@ -623,56 +505,21 @@ func (conn *NetConnection) readChunkType(h *ChunkHeader, chunkType byte) (head *
}
}
return h, nil
return nil
}
func (conn *NetConnection) RecvMessage() (msg *Chunk, err error) {
if conn.readSeqNum >= conn.bandwidth {
conn.totalRead += conn.readSeqNum
conn.readSeqNum = 0
//sendAck(conn, conn.totalRead)
conn.SendMessage(SEND_ACK_MESSAGE, conn.totalRead)
err = conn.SendMessage(RTMP_MSG_ACK, Uint32Message(conn.totalRead))
}
if msg, err = conn.readChunk(); err != nil {
return nil, err
for msg == nil && err == nil {
msg, err = conn.readChunk()
}
// 如果消息是类型是用户控制消息,那么我们就简单做一些相应的处理,
// 然后继续读取下一个消息.如果不是用户控制消息,就将消息返回就好.
messageType := msg.MessageTypeID
if RTMP_MSG_CHUNK_SIZE <= messageType && messageType <= RTMP_MSG_EDGE {
switch messageType {
case RTMP_MSG_CHUNK_SIZE:
m := msg.MsgData.(Uint32Message)
conn.readChunkSize = int(m)
return conn.RecvMessage()
case RTMP_MSG_ABORT:
m := msg.MsgData.(Uint32Message)
delete(conn.incompleteRtmpBody, uint32(m))
return conn.RecvMessage()
case RTMP_MSG_ACK, RTMP_MSG_EDGE:
return conn.RecvMessage()
case RTMP_MSG_USER_CONTROL:
if _, ok := msg.MsgData.(*PingRequestMessage); ok {
//sendPingResponse(conn)
conn.SendMessage(SEND_PING_RESPONSE_MESSAGE, nil)
return
}
return conn.RecvMessage()
case RTMP_MSG_ACK_SIZE:
m := msg.MsgData.(Uint32Message)
conn.bandwidth = uint32(m)
return conn.RecvMessage()
case RTMP_MSG_BANDWIDTH:
m := msg.MsgData.(*SetPeerBandwidthMessage)
conn.bandwidth = m.AcknowledgementWindowsize
return conn.RecvMessage()
}
}
return msg, err
}
func (conn *NetConnection) writeMessage(t byte, msg RtmpMessage) error {
func (conn *NetConnection) SendMessage(t byte, msg RtmpMessage) (err error) {
body := msg.Encode()
head := newChunkHeader(t)
head.MessageLength = uint32(len(body))
@@ -682,24 +529,29 @@ func (conn *NetConnection) writeMessage(t byte, msg RtmpMessage) error {
if conn.writeSeqNum > conn.bandwidth {
conn.totalWrite += conn.writeSeqNum
conn.writeSeqNum = 0
conn.SendMessage(SEND_ACK_MESSAGE, conn.totalWrite)
conn.SendMessage(SEND_PING_REQUEST_MESSAGE, nil)
err = conn.SendMessage(RTMP_MSG_ACK, Uint32Message(conn.totalWrite))
err = conn.SendStreamID0(RTMP_USER_PING_REQUEST)
}
var chunk = net.Buffers{conn.encodeChunk12(head)}
if len(body) > conn.writeChunkSize {
chunk = append(chunk, body[:conn.writeChunkSize])
body = body[conn.writeChunkSize:]
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk))
_, err = chunk.WriteTo(conn)
for len(body) > conn.writeChunkSize {
chunk = append(chunk[:0], conn.encodeChunk12(head), body[:conn.writeChunkSize])
body = body[conn.writeChunkSize:]
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk))
_, err = chunk.WriteTo(conn)
}
chunk = append(chunk[:0], conn.encodeChunk12(head), body)
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk))
_, err = chunk.WriteTo(conn)
} else {
chunk = append(chunk, body)
conn.writeSeqNum += uint32(util.SizeOfBuffers(chunk))
_, err = chunk.WriteTo(conn)
}
need, err := conn.encodeChunk12(head, body)
if err != nil {
return err
}
if err = conn.Flush(); err != nil {
return err
}
for need != nil && len(need) > 0 {
if need, err = conn.encodeChunk1(head, need); err != nil {
return err
}
if err = conn.Flush(); err != nil {
return err
}
}
return nil
}

177
netStream.go
View File

@@ -5,43 +5,30 @@ import (
"fmt"
"log"
"net"
"strings"
"sync/atomic"
"github.com/Monibuca/engine/v3"
"github.com/Monibuca/utils/v3"
"github.com/Monibuca/engine/v4"
"github.com/Monibuca/engine/v4/codec"
"github.com/Monibuca/engine/v4/util"
)
var gstreamid = uint32(64)
func processRtmp(conn net.Conn) {
var stream *engine.Stream
streams := make(map[uint32]*engine.Subscriber)
defer func() {
conn.Close()
if stream != nil {
stream.Close()
}
for _, s := range streams {
s.Close()
}
}()
func (cfg *RTMPConfig) Process(conn *net.TCPConn) {
nc := NetConnection{
ReadWriter: bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)),
TCPConn: conn,
Reader: bufio.NewReader(conn),
writeChunkSize: RTMP_DEFAULT_CHUNK_SIZE,
readChunkSize: RTMP_DEFAULT_CHUNK_SIZE,
rtmpHeader: make(map[uint32]*ChunkHeader),
incompleteRtmpBody: make(map[uint32][]byte),
incompleteRtmpBody: make(map[uint32]util.Buffer),
bandwidth: RTMP_MAX_CHUNK_SIZE << 3,
nextStreamID: func() uint32 {
return atomic.AddUint32(&gstreamid, 1)
},
tmpBuf: make([]byte, 4),
subscribers: make(map[uint32]*engine.Subscriber),
}
defer nc.Close()
/* Handshake */
if utils.MayBeError(Handshake(nc.ReadWriter)) {
return
}
if utils.MayBeError(nc.OnConnect()) {
if util.MayBeError(nc.Handshake()) {
return
}
var rec_audio, rec_video func(*Chunk)
@@ -52,27 +39,59 @@ func processRtmp(conn net.Conn) {
continue
}
switch msg.MessageTypeID {
case RTMP_MSG_CHUNK_SIZE:
nc.readChunkSize = int(msg.MsgData.(Uint32Message))
case RTMP_MSG_ABORT:
delete(nc.incompleteRtmpBody, uint32(msg.MsgData.(Uint32Message)))
case RTMP_MSG_ACK, RTMP_MSG_EDGE:
case RTMP_MSG_USER_CONTROL:
if _, ok := msg.MsgData.(*PingRequestMessage); ok {
nc.SendUserControl(RTMP_USER_PING_RESPONSE)
}
case RTMP_MSG_ACK_SIZE:
nc.bandwidth = uint32(msg.MsgData.(Uint32Message))
case RTMP_MSG_BANDWIDTH:
m := msg.MsgData.(*SetPeerBandwidthMessage)
nc.bandwidth = m.AcknowledgementWindowsize
case RTMP_MSG_AMF0_COMMAND:
if msg.MsgData == nil {
break
}
cmd := msg.MsgData.(Commander).GetCommand()
util.Println(cmd.CommandName)
switch cmd.CommandName {
case "connect":
connect := msg.MsgData.(*CallMessage)
app := connect.Object["app"] // 客户端要连接到的服务应用名
objectEncoding := connect.Object["objectEncoding"] // AMF编码方法
if objectEncoding != nil {
nc.objectEncoding = objectEncoding.(float64)
}
nc.appName = app.(string)
log.Printf("app:%v,objectEncoding:%v", app, objectEncoding)
err = nc.SendMessage(RTMP_MSG_ACK_SIZE, Uint32Message(512<<10))
nc.writeChunkSize = config.ChunkSize
err = nc.SendMessage(RTMP_MSG_CHUNK_SIZE, Uint32Message(config.ChunkSize))
err = nc.SendMessage(RTMP_MSG_BANDWIDTH, &SetPeerBandwidthMessage{
AcknowledgementWindowsize: uint32(512 << 10),
LimitType: byte(2),
})
err = nc.SendStreamID(RTMP_USER_STREAM_BEGIN)
err = nc.SendCommand(SEND_CONNECT_RESPONSE_MESSAGE, nc.objectEncoding)
case "createStream":
nc.streamID = nc.nextStreamID()
nc.streamID = atomic.AddUint32(&gstreamid, 1)
log.Println("createStream:", nc.streamID)
err = nc.SendMessage(SEND_CREATE_STREAM_RESPONSE_MESSAGE, cmd.TransactionId)
if utils.MayBeError(err) {
err = nc.SendCommand(SEND_CREATE_STREAM_RESPONSE_MESSAGE, cmd.TransactionId)
if util.MayBeError(err) {
return
}
case "publish":
pm := msg.MsgData.(*PublishMessage)
streamPath := nc.appName + "/" + strings.Split(pm.PublishingName, "?")[0]
stream = &engine.Stream{Type: "RTMP", StreamPath: streamPath}
if stream.Publish() {
nc.Config = config.Publish
if nc.Publish(nc.appName+"/"+pm.PublishingName, &nc) {
absTs := make(map[uint32]uint32)
vt := stream.NewVideoTrack(0)
at := stream.NewAudioTrack(0)
vt := nc.Stream.NewVideoTrack()
at := nc.Stream.NewAudioTrack()
rec_audio = func(msg *Chunk) {
if msg.ChunkType == 0 {
absTs[msg.ChunkStreamID] = 0
@@ -82,7 +101,7 @@ func processRtmp(conn net.Conn) {
} else {
absTs[msg.ChunkStreamID] += msg.Timestamp
}
at.PushByteStream(absTs[msg.ChunkStreamID], msg.Body)
at.WriteAVCC(absTs[msg.ChunkStreamID], msg.Body)
}
rec_video = func(msg *Chunk) {
if msg.ChunkType == 0 {
@@ -93,99 +112,79 @@ func processRtmp(conn net.Conn) {
} else {
absTs[msg.ChunkStreamID] += msg.Timestamp
}
vt.PushByteStream(absTs[msg.ChunkStreamID], msg.Body)
vt.WriteAVCC(absTs[msg.ChunkStreamID], 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))
nc.SendStreamID(RTMP_USER_STREAM_BEGIN)
err = nc.SendCommand(SEND_PUBLISH_START_MESSAGE, newPublishResponseMessageData(nc.streamID, NetStream_Publish_Start, Level_Status))
} else {
err = nc.SendMessage(SEND_PUBLISH_RESPONSE_MESSAGE, newPublishResponseMessageData(nc.streamID, NetStream_Publish_BadName, Level_Error))
err = nc.SendCommand(SEND_PUBLISH_RESPONSE_MESSAGE, newPublishResponseMessageData(nc.streamID, NetStream_Publish_BadName, Level_Error))
}
case "play":
pm := msg.MsgData.(*PlayMessage)
streamPath := nc.appName + "/" + strings.Split(pm.StreamName, "?")[0]
nc.writeChunkSize = config.ChunkSize
streamPath := nc.appName + "/" + pm.StreamName
subscriber := engine.Subscriber{
Type: "RTMP",
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_STREAM_IS_RECORDED_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_Start, Level_Status))
vt, at := subscriber.WaitVideoTrack(), subscriber.WaitAudioTrack()
if subscriber.Subscribe(streamPath, config.Subscribe) {
nc.subscribers[nc.streamID] = &subscriber
err = nc.SendStreamID(RTMP_USER_STREAM_IS_RECORDED)
err = nc.SendStreamID(RTMP_USER_STREAM_BEGIN)
err = nc.SendCommand(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Reset, Level_Status))
err = nc.SendCommand(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Start, Level_Status))
vt, at := subscriber.WaitVideoTrack("h264", "h265"), subscriber.WaitAudioTrack("aac", "pcma", "pcmu")
if vt != nil {
var lastTimeStamp 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
}
err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.ExtraData.Payload})
subscriber.OnVideo = func(ts uint32, pack *engine.VideoPack) {
err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Timestamp: 0, Payload: pack.Payload})
subscriber.OnVideo = func(ts uint32, pack *engine.VideoPack) {
err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: getDeltaTime(ts), Payload: pack.Payload})
}
frame := vt.DecoderConfiguration
err = nc.sendAVMessage(0, frame.AVCC, false, true)
subscriber.OnVideo = func(frame *engine.VideoFrame) bool {
err = nc.sendAVMessage(frame.DeltaTime, frame.AVCC, false, false)
return err == nil
}
}
if at != nil {
var lastTimeStamp 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
subscriber.OnAudio = func(frame *engine.AudioFrame) bool {
if at.CodecID == codec.CodecID_AAC {
frame := at.DecoderConfiguration
err = nc.sendAVMessage(0, frame.AVCC, true, true)
} else {
err = nc.sendAVMessage(0, frame.AVCC, true, true)
}
return
subscriber.OnAudio = func(frame *engine.AudioFrame) bool {
err = nc.sendAVMessage(frame.DeltaTime, frame.AVCC, true, false)
return err == nil
}
subscriber.OnAudio = func(ts uint32, pack *engine.AudioPack) {
if at.CodecID == 10 {
err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.ExtraData})
}
subscriber.OnAudio = func(ts uint32, pack *engine.AudioPack) {
err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: getDeltaTime(ts), Payload: pack.Payload})
}
subscriber.OnAudio(ts, pack)
return err == nil
}
}
go subscriber.Play(at, vt)
} else {
err = nc.SendCommand(SEND_PLAY_RESPONSE_MESSAGE, newPlayResponseMessageData(nc.streamID, NetStream_Play_Failed, Level_Error))
}
case "closeStream":
cm := msg.MsgData.(*CURDStreamMessage)
if stream, ok := streams[cm.StreamId]; ok {
if stream, ok := nc.subscribers[cm.StreamId]; ok {
stream.Close()
delete(streams, cm.StreamId)
delete(nc.subscribers, cm.StreamId)
}
case "releaseStream":
cm := msg.MsgData.(*ReleaseStreamMessage)
streamPath := nc.appName + "/" + strings.Split(cm.StreamName, "?")[0]
amfobj := newAMFObjects()
if s := engine.FindStream(streamPath); s != nil {
amfobj := make(AMFObject)
if nc.Stream != nil && nc.Stream.AppName == nc.appName && nc.Stream.StreamName == cm.StreamName {
amfobj["level"] = "_result"
s.Close()
nc.Stream.UnPublish()
} else {
amfobj["level"] = "_error"
}
amfobj["tid"] = cm.TransactionId
err = nc.SendMessage(SEND_UNPUBLISH_RESPONSE_MESSAGE, amfobj)
err = nc.SendCommand(SEND_UNPUBLISH_RESPONSE_MESSAGE, amfobj)
}
case RTMP_MSG_AUDIO:
rec_audio(msg)
case RTMP_MSG_VIDEO:
rec_video(msg)
}
msg.Recycle()
} else {
util.Println(err)
return
}
}