engine

package module
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 18, 2021 License: MIT Imports: 18 Imported by: 0

README

m7s核心引擎

该项目为m7s的引擎部分，该部分逻辑是流媒体服务器的核心转发逻辑。仅包含最基础的功能，不含任何网络协议部分，但包含了一个插件的引入机制，其他功能均由插件实现

引擎的基本功能

  • 引擎初始化会加载配置文件，并逐个调用插件的Run函数
  • 具有发布功能的插件，会通过GetStream函数创建一个流，即Stream对象，这个Stream对象随后可以被订阅
  • Stream对象中含有两个列表，一个是VideoTracks一个是AudioTracks用来存放视频数据和音频数据
  • 每一个VideoTrack或者AudioTrack中包含一个RingBuffer，用来存储发布者提供的数据，同时提供订阅者访问。
  • 具有订阅功能的插件，会通过GetStream函数获取到一个流，然后选择VideoTracks、AudioTracks里面的RingBuffer进行连续的读取

发布插件如何发布流

以rtmp协议为例子

if pub := new(engine.Publisher); pub.Publish(streamPath) {
    pub.Type = "RTMP"
    stream = pub.Stream
    err = nc.SendMessage(SEND_STREAM_BEGIN_MESSAGE, nil)
    err = nc.SendMessage(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))
}

默认会创建一个VideoTrack和一个AudioTrack 当我们接收到数据的时候就可以朝里面填充物数据了

rec_video = func(msg *Chunk) {
    // 等待AVC序列帧
    if msg.Body[1] != 0 {
        return
    }
    vt := stream.VideoTracks[0]
    var ts_video uint32
    var info codec.AVCDecoderConfigurationRecord
    //0:codec,1:IsAVCSequence,2~4:compositionTime
    if _, err := info.Unmarshal(msg.Body[5:]); err == nil {
        vt.SPSInfo, err = codec.ParseSPS(info.SequenceParameterSetNALUnit)
        vt.SPS = info.SequenceParameterSetNALUnit
        vt.PPS = info.PictureParameterSetNALUnit
    }
    vt.RtmpTag = msg.Body
    nalulenSize := int(info.LengthSizeMinusOne&3 + 1)
    rec_video = func(msg *Chunk) {
        nalus := msg.Body[5:]
        if msg.Timestamp == 0xffffff {
            ts_video += msg.ExtendTimestamp
        } else {
            ts_video += msg.Timestamp // 绝对时间戳
        }
        for len(nalus) > nalulenSize {
            nalulen := 0
            for i := 0; i < nalulenSize; i++ {
                nalulen += int(nalus[i]) << (8 * (nalulenSize - i - 1))
            }
            vt.Push(ts_video, nalus[nalulenSize:nalulen+nalulenSize])
            nalus = nalus[nalulen+nalulenSize:]
        }
    }
    close(vt.WaitFirst)
}

在填充数据之前，需要获取到SPS和PPS，然后设置好，因为订阅者需要先发送这个数据 然后通过Track到Push函数将数据填充到RingBuffer里面去

订阅插件如何订阅流

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.GetVideoTrack("h264"), subscriber.OriginAudioTrack
    if vt != nil {
        var lastVideoTime uint32
        err = nc.SendMessage(SEND_FULL_VDIEO_MESSAGE, &AVPack{Payload: vt.RtmpTag})
        subscriber.OnVideo = func(pack engine.VideoPack) {
            if lastVideoTime == 0 {
                lastVideoTime = pack.Timestamp
            }
            t := pack.Timestamp - lastVideoTime
            lastVideoTime = pack.Timestamp
            payload := codec.Nalu2RTMPTag(pack.Payload)
            defer utils.RecycleSlice(payload)
            err = nc.SendMessage(SEND_VIDEO_MESSAGE, &AVPack{Timestamp: t, Payload: payload})
        }
    }
    if at != nil {
        var lastAudioTime uint32
        var aac byte
        if at.SoundFormat == 10 {
            aac = at.RtmpTag[0]
            err = nc.SendMessage(SEND_FULL_AUDIO_MESSAGE, &AVPack{Payload: at.RtmpTag})
        }
        subscriber.OnAudio = func(pack engine.AudioPack) {
            if lastAudioTime == 0 {
                lastAudioTime = pack.Timestamp
            }
            t := pack.Timestamp - lastAudioTime
            lastAudioTime = pack.Timestamp
            payload := codec.Audio2RTMPTag(aac, pack.Payload)
            defer utils.RecycleSlice(payload)
            err = nc.SendMessage(SEND_AUDIO_MESSAGE, &AVPack{Timestamp: t, Payload: payload})
        }
    }
    go subscriber.Play(at, vt)
}
  • 在发送数据前，需要先发送音视频的序列帧

Documentation

Index

Constants

View Source 
const (
	HOOK_SUBSCRIBE          = "Subscribe"
	HOOK_UNSUBSCRIBE        = "UnSubscibe"
	HOOK_STREAMCLOSE        = "StreamClose"
	HOOK_PUBLISH            = "Publish"
	HOOK_REQUEST_TRANSAUDIO = "RequestTransAudio"
)

Variables

View Source 
var Hooks = make(map[string]*RingBuffer)

Functions

func AddHook 

func AddHook(name string, callback interface{})

func AddHookConditional 

func AddHookConditional(name string, callback interface{}, goon func() bool)

func AddHooks 

func AddHooks(hooks map[string]interface{})

func TriggerHook 

func TriggerHook(name string, payload ...interface{})

Types

type AVItem 

type AVItem struct {
	DataItem
	// contains filtered or unexported fields
}

type AVPack 

type AVPack struct {
	bytes.Buffer
	Payload []byte // 字节流格式的媒体数据，如果是需要拼接而成的，则等同于Buffer里面的值
}

func (*AVPack) Bytes2Payload 

func (pack *AVPack) Bytes2Payload()

type AVRing 

type AVRing struct {
	RingBuffer
	// contains filtered or unexported fields
}

func (AVRing) Clone 

func (r AVRing) Clone() *AVRing

func (*AVRing) Current 

func (r *AVRing) Current() *AVItem

func (*AVRing) CurrentValue 

func (r *AVRing) CurrentValue() interface{}

func (*AVRing) GetNext 

func (r *AVRing) GetNext() *AVItem

func (*AVRing) Init 

func (r *AVRing) Init(ctx context.Context, n int) *AVRing

func (*AVRing) NextRead 

func (r *AVRing) NextRead() (item *AVItem, value interface{})

func (*AVRing) NextValue 

func (r *AVRing) NextValue() interface{}

func (*AVRing) PreItem 

func (r *AVRing) PreItem() *AVItem

func (*AVRing) Read 

func (r *AVRing) Read() (item *AVItem, value interface{})

func (*AVRing) Step 

func (r *AVRing) Step()

func (AVRing) SubRing 

func (r AVRing) SubRing(rr *ring.Ring) *AVRing

func (*AVRing) Write 

func (r *AVRing) Write(value interface{})

type AVTrack 

type AVTrack struct {
	AVRing  `json:"-"`
	CodecID byte
	BaseTrack
	*AVItem `json:"-"` //当前正在写入的数据对象
	// contains filtered or unexported fields
}

func (*AVTrack) GetBPS 

func (t *AVTrack) GetBPS()

type AudioPack 

type AudioPack struct {
	AVPack
	Raw []byte
}

type AudioTrack 

type AudioTrack struct {
	AVTrack
	SoundRate      int                             //2bit
	SoundSize      byte                            //1bit
	Channels       byte                            //1bit
	ExtraData      []byte                          `json:"-"` //rtmp协议需要先发这个帧
	PushByteStream func(ts uint32, payload []byte) `json:"-"`
	PushRaw        func(ts uint32, payload []byte) `json:"-"`

	*AudioPack `json:"-"` // 当前正在写入的音频对象
	// contains filtered or unexported fields
}

func (*AudioTrack) Play 

func (at *AudioTrack) Play(onAudio func(uint32, *AudioPack), exit1, exit2 <-chan struct{})

func (*AudioTrack) SetASC 

func (at *AudioTrack) SetASC(asc []byte)

type BaseTrack 

type BaseTrack struct {
	Stream      *Stream `json:"-"`
	PacketCount int
	BPS         int
	// contains filtered or unexported fields
}

type DataItem 

type DataItem struct {
	Timestamp time.Time
	Sequence  int
	Value     interface{}
}

type DataTrack 

type DataTrack struct {
	RingBuffer
	BaseTrack
	*LockItem
	sync.Locker // 写入锁，可选，单一写入可以不加锁
}

func (*DataTrack) GetBPS 

func (t *DataTrack) GetBPS()

func (*DataTrack) Play 

func (dt *DataTrack) Play(onData func(*DataItem), exit1, exit2 <-chan struct{})

func (*DataTrack) Push 

func (dt *DataTrack) Push(data interface{})

type LockItem 

type LockItem struct {
	DataItem
	sync.RWMutex
}

type RTPAudio 

type RTPAudio struct {
	RTPPublisher
	*AudioTrack
}

type RTPNalu 

type RTPNalu struct {
	Payload []byte
	Ts      uint32
	Next    *RTPNalu
}

type RTPPublisher 

type RTPPublisher struct {
	rtp.Packet `json:"-"`
	// contains filtered or unexported fields
}

func (*RTPPublisher) Push 

func (p *RTPPublisher) Push(payload []byte)

type RTPVideo 

type RTPVideo struct {
	RTPPublisher
	*VideoTrack
	// contains filtered or unexported fields
}

type RingBuffer 

type RingBuffer struct {
	*ring.Ring
	Size int
	Flag *int32
	context.Context
}

func (RingBuffer) Clone 

func (rb RingBuffer) Clone() *RingBuffer

func (*RingBuffer) Current 

func (rb *RingBuffer) Current() *LockItem

func (*RingBuffer) CurrentValue 

func (rb *RingBuffer) CurrentValue() interface{}

func (*RingBuffer) Dispose 

func (rb *RingBuffer) Dispose()

func (*RingBuffer) GetNext 

func (rb *RingBuffer) GetNext() *LockItem

func (*RingBuffer) Init 

func (rb *RingBuffer) Init(ctx context.Context, n int) *RingBuffer

func (*RingBuffer) MoveNext 

func (rb *RingBuffer) MoveNext()

func (*RingBuffer) NextValue 

func (rb *RingBuffer) NextValue() interface{}

func (*RingBuffer) Read 

func (rb *RingBuffer) Read() interface{}

func (*RingBuffer) ReadLoop 

func (rb *RingBuffer) ReadLoop(handler interface{})

ReadLoop 循环读取，采用了反射机制，不适用高性能场景 handler入参可以传入回调函数或者channel

func (*RingBuffer) ReadLoopConditional 

func (rb *RingBuffer) ReadLoopConditional(handler interface{}, goon func() bool)

goon判断函数用来判断是否继续读取,返回false将终止循环

func (*RingBuffer) Step 

func (rb *RingBuffer) Step()

func (RingBuffer) SubRing 

func (rb RingBuffer) SubRing(rr *ring.Ring) *RingBuffer

func (*RingBuffer) Write 

func (rb *RingBuffer) Write(value interface{})

type Stream 

type Stream struct {
	context.Context `json:"-"`
	StreamPath      string
	Type            string        //流类型，来自发布者
	StartTime       time.Time     //流的创建时间
	Subscribers     []*Subscriber // 订阅者
	VideoTracks     Tracks
	AudioTracks     Tracks
	DataTracks      Tracks
	AutoUnPublish   bool              //	当无人订阅时自动停止发布
	Transcoding     map[string]string //转码配置，key：目标编码，value：发布者提供的编码

	OnClose   func()      `json:"-"`
	ExtraProp interface{} //额外的属性，用于实现子类化，减少map的使用
	// contains filtered or unexported fields
}

Stream 流定义

func FindStream 

func FindStream(streamPath string) *Stream

FindStream 根据流路径查找流

func Publish 

func Publish(streamPath, t string) *Stream

Publish 直接发布

func (*Stream) AddOnClose 

func (r *Stream) AddOnClose(onClose func())

增加结束时的回调，使用类似Js的猴子补丁

func (*Stream) Close 

func (r *Stream) Close()

func (*Stream) NewAudioTrack 

func (s *Stream) NewAudioTrack(codec byte) (at *AudioTrack)

func (*Stream) NewDataTrack 

func (s *Stream) NewDataTrack(l sync.Locker) (dt *DataTrack)

func (*Stream) NewRTPAudio 

func (s *Stream) NewRTPAudio(codec byte) (r *RTPAudio)

func (*Stream) NewRTPVideo 

func (s *Stream) NewRTPVideo(codec byte) (r *RTPVideo)

func (*Stream) NewVideoTrack 

func (s *Stream) NewVideoTrack(codec byte) (vt *VideoTrack)

func (*Stream) Publish 

func (r *Stream) Publish() bool

Publish 发布者进行发布操作

func (*Stream) Subscribe 

func (r *Stream) Subscribe(s *Subscriber)

Subscribe 订阅流

func (*Stream) UnSubscribe 

func (r *Stream) UnSubscribe(s *Subscriber)

UnSubscribe 取消订阅流

func (*Stream) Update 

func (r *Stream) Update()

func (*Stream) WaitAudioTrack 

func (r *Stream) WaitAudioTrack(names ...string) *AudioTrack

TODO: 触发转码逻辑

func (*Stream) WaitDataTrack 

func (r *Stream) WaitDataTrack(names ...string) *DataTrack

func (*Stream) WaitVideoTrack 

func (r *Stream) WaitVideoTrack(names ...string) *VideoTrack

type StreamCollection 

type StreamCollection struct {
	sync.RWMutex
	// contains filtered or unexported fields
}
var Streams StreamCollection

Streams 所有的流集合

func (*StreamCollection) Delete 

func (sc *StreamCollection) Delete(streamPath string)

func (*StreamCollection) GetStream 

func (sc *StreamCollection) GetStream(streamPath string) *Stream

func (*StreamCollection) Range 

func (sc *StreamCollection) Range(f func(*Stream))

func (*StreamCollection) ToList 

func (sc *StreamCollection) ToList() (r []*Stream)

type Subscriber 

type Subscriber struct {
	context.Context `json:"-"`

	Ctx2          context.Context `json:"-"`
	*Stream       `json:"-"`
	ID            string
	TotalDrop     int //总丢帧
	TotalPacket   int
	Type          string
	BufferLength  int
	Delay         uint32
	SubscribeTime time.Time
	SubscribeArgs url.Values
	OnAudio       func(uint32, *AudioPack) `json:"-"`
	OnVideo       func(uint32, *VideoPack) `json:"-"`
	// contains filtered or unexported fields
}

Subscriber 订阅者实体定义

func DeleteSliceItem_Subscriber 

func DeleteSliceItem_Subscriber(slice []*Subscriber, item *Subscriber) ([]*Subscriber, bool)

func (*Subscriber) Close 

func (s *Subscriber) Close()

Close 关闭订阅者

func (*Subscriber) Play 

func (s *Subscriber) Play(at *AudioTrack, vt *VideoTrack)

Play 开始播放

func (*Subscriber) PlayAudio 

func (s *Subscriber) PlayAudio(at *AudioTrack)

func (*Subscriber) PlayVideo 

func (s *Subscriber) PlayVideo(vt *VideoTrack)

func (*Subscriber) Subscribe 

func (s *Subscriber) Subscribe(streamPath string) error

Subscribe 开始订阅 将Subscriber与Stream关联

type TSSlice 

type TSSlice []uint32

func (TSSlice) Len 

func (s TSSlice) Len() int

func (TSSlice) Less 

func (s TSSlice) Less(i, j int) bool

func (TSSlice) Swap 

func (s TSSlice) Swap(i, j int)

type Track 

type Track interface {
	GetBPS()
}

type Tracks 

type Tracks struct {
	RingBuffer

	context.Context
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*Tracks) AddTrack 

func (ts *Tracks) AddTrack(name string, t Track)

func (*Tracks) GetTrack 

func (ts *Tracks) GetTrack(name string) Track

func (*Tracks) Init 

func (ts *Tracks) Init(ctx context.Context)

func (*Tracks) MarshalJSON 

func (ts *Tracks) MarshalJSON() ([]byte, error)

func (*Tracks) OnTrack 

func (ts *Tracks) OnTrack(callback func(string, Track))

func (*Tracks) WaitTrack 

func (ts *Tracks) WaitTrack(names ...string) Track

type TransCodeReq 

type TransCodeReq struct {
	*Subscriber
	RequestCodec string
}

type VideoPack 

type VideoPack struct {
	AVPack
	CompositionTime uint32
	NALUs           [][]byte
	IDR             bool // 是否关键帧
}

func (*VideoPack) ResetNALUs 

func (v *VideoPack) ResetNALUs()

func (*VideoPack) SetNalu0 

func (v *VideoPack) SetNalu0(nalu []byte)

type VideoTrack 

type VideoTrack struct {
	IDRing *ring.Ring `json:"-"` //最近的关键帧位置，首屏渲染
	AVTrack
	SPSInfo   codec.SPSInfo
	GOP       int           //关键帧间隔
	ExtraData *VideoPack    `json:"-"` //H264(SPS、PPS) H265(VPS、SPS、PPS)
	WaitIDR   chan struct{} `json:"-"`

	PushNalu       func(ts uint32, cts uint32, nalus ...[]byte) `json:"-"`
	UsingDonlField bool

	*VideoPack `json:"-"` //当前写入的视频数据
	// contains filtered or unexported fields
}

func (*VideoTrack) Play 

func (vt *VideoTrack) Play(onVideo func(uint32, *VideoPack), exit1, exit2 <-chan struct{})

func (*VideoTrack) PushAnnexB 

func (vt *VideoTrack) PushAnnexB(ts uint32, cts uint32, payload []byte)

func (*VideoTrack) PushByteStream 

func (vt *VideoTrack) PushByteStream(ts uint32, payload []byte)

Source Files

View all Source files

Directories

Path Synopsis
mpegts

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.