engine

package module
v3.4.4 Latest Latest
Warning

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

 Go to latest Published: Jan 5, 2022 License: MIT Imports: 28 Imported by: 28

README

m7s核心引擎

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

引擎配置

[Engine]
EnableAudio = true
EnableVideo = true
# 发布流默认过期时间单位秒
PublishTimeout = 60
# 自动关闭触发后延迟的秒数(期间内如果有新的订阅则取消触发关闭)
AutoCloseDelay = 10
# 启用RTP包乱序重排
RTPReorder = false

引擎的基本功能

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

发布插件如何发布流

以rtmp协议为例子

stream = &engine.Stream{Type: "RTMP", StreamPath: streamPath}
if stream.Publish() {
  absTs := make(map[uint32]uint32)
  vt := stream.NewVideoTrack(0)
  at := stream.NewAudioTrack(0)
  rec_audio = func(msg *Chunk) {
    if msg.ChunkType == 0 {
      absTs[msg.ChunkStreamID] = 0
    }
    if msg.Timestamp == 0xffffff {
      absTs[msg.ChunkStreamID] += msg.ExtendTimestamp
    } else {
      absTs[msg.ChunkStreamID] += msg.Timestamp
    }
    at.PushByteStream(absTs[msg.ChunkStreamID], msg.Body)
  }
  rec_video = func(msg *Chunk) {
    if msg.ChunkType == 0 {
      absTs[msg.ChunkStreamID] = 0
    }
    if msg.Timestamp == 0xffffff {
      absTs[msg.ChunkStreamID] += msg.ExtendTimestamp
    } else {
      absTs[msg.ChunkStreamID] += msg.Timestamp
    }
    vt.PushByteStream(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))
} else {
  err = nc.SendMessage(SEND_PUBLISH_RESPONSE_MESSAGE, newPublishResponseMessageData(nc.streamID, NetStream_Publish_BadName, Level_Error))
}

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

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

订阅插件如何订阅流

sub := Subscriber{ID: r.RemoteAddr, Type: "FLV", Ctx2: r.Context()}
if err := sub.Subscribe(stringPath); err == nil {
  vt, at := sub.WaitVideoTrack(), sub.WaitAudioTrack()
  var buffer bytes.Buffer
  if _, err := amf.WriteString(&buffer, "onMetaData"); err != nil {
    return
  }
  if vt != nil {
    codec.WriteFLVTag(w, codec.FLV_TAG_TYPE_VIDEO, 0, vt.ExtraData.Payload)
    sub.OnVideo = func(ts uint32, pack *VideoPack) {
      codec.WriteFLVTag(w, codec.FLV_TAG_TYPE_VIDEO, ts, pack.Payload)
    }
  }
  if at != nil {
    if at.CodecID == 10 {
      codec.WriteFLVTag(w, codec.FLV_TAG_TYPE_AUDIO, 0, at.ExtraData)
    }
    sub.OnAudio = func(ts uint32, pack *AudioPack) {
      codec.WriteFLVTag(w, codec.FLV_TAG_TYPE_AUDIO, ts, pack.Payload)
    }
  }
  sub.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 (

	// ConfigRaw 配置信息的原始数据
	ConfigRaw     []byte
	StartTime     time.Time                        //启动时间
	Plugins       = make(map[string]*PluginConfig) // Plugins 所有的插件配置
	HasTranscoder bool
	Ctx           context.Context
)
View Source 
var Hooks = make(map[string]*RingBuffer)
View Source 
var StreamTimeoutError = errors.New("timeout")
View Source 
var Version = "3.2.2"

Functions

func AddHook 

func AddHook(name string, callback interface{})

func AddHookConditional added in v3.1.8 

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

func AddHookGo added in v3.3.14 

func AddHookGo(name string, callback interface{})

func AddHooks added in v3.1.5 

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

func InstallPlugin 

func InstallPlugin(opt *PluginConfig)

InstallPlugin 安装插件

func Run 

func Run(ctx context.Context, configFile string) (err error)

Run 启动Monibuca引擎

func TriggerHook 

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

Types

type AVItem added in v3.2.5 

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

type AVPack added in v3.1.0 

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

func (*AVPack) Bytes2Payload added in v3.3.1 

func (pack *AVPack) Bytes2Payload()

type AVRing added in v3.2.5 

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

func (AVRing) Clone added in v3.2.5 

func (r AVRing) Clone() *AVRing

func (*AVRing) Current added in v3.2.5 

func (r *AVRing) Current() *AVItem

func (*AVRing) CurrentValue added in v3.2.5 

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

func (*AVRing) GetNext added in v3.2.5 

func (r *AVRing) GetNext() *AVItem

func (*AVRing) Init added in v3.2.5 

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

func (*AVRing) NextValue added in v3.2.5 

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

func (*AVRing) PreItem added in v3.3.2 

func (r *AVRing) PreItem() *AVItem

func (*AVRing) Read added in v3.2.5 

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

func (*AVRing) Step added in v3.2.5 

func (r *AVRing) Step()

func (AVRing) SubRing added in v3.2.5 

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

func (*AVRing) TryRead added in v3.3.10 

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

func (*AVRing) Write added in v3.2.5 

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

type AVTrack added in v3.3.1 

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

func (*AVTrack) GetBPS added in v3.3.1 

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 added in v3.1.3 

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

func (*AudioTrack) SetASC 

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

type B added in v3.3.11 

type B struct {
	TSSlice

	MaxTS uint32
	// contains filtered or unexported fields
}

func (*B) Pop added in v3.3.11 

func (b *B) Pop() interface{}

func (*B) Push added in v3.3.11 

func (b *B) Push(x interface{})

type BaseTrack added in v3.3.1 

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

type DTSEstimator added in v3.3.12 

type DTSEstimator struct {
	// contains filtered or unexported fields
}

DTSEstimator is a DTS estimator.

func NewDTSEstimator added in v3.3.12 

func NewDTSEstimator() *DTSEstimator

NewDTSEstimator allocates a DTSEstimator.

func (*DTSEstimator) Feed added in v3.3.12 

func (d *DTSEstimator) Feed(pts uint32) uint32

Feed provides PTS to the estimator, and returns the estimated DTS.

type DataItem added in v3.3.1 

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

type DataTrack added in v3.3.1 

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

func (*DataTrack) GetBPS added in v3.3.1 

func (t *DataTrack) GetBPS()

func (*DataTrack) Play added in v3.3.1 

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

func (*DataTrack) Push added in v3.3.1 

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

type LockItem added in v3.2.5 

type LockItem struct {
	DataItem
	sync.RWMutex
}

type PluginConfig 

type PluginConfig struct {
	Name      string                       //插件名称
	Config    interface{}                  //插件配置
	Version   string                       //插件版本
	Dir       string                       //插件代码路径
	Run       func()                       //插件启动函数
	HotConfig map[string]func(interface{}) //热修改配置
}

PluginConfig 插件配置定义

func (*PluginConfig) Install added in v3.4.0 

func (opt *PluginConfig) Install(run func())

func (*PluginConfig) Save added in v3.4.0 

func (opt *PluginConfig) Save() error

type RTPAudio 

type RTPAudio struct {
	RTPDemuxer `json:"-"`
	*AudioTrack
}

type RTPDemuxer added in v3.3.12 

type RTPDemuxer struct {
	rtp.Packet
	PTS uint32 // 修复后的时间戳（毫秒）

	OnDemux func(uint32, []byte)
	// contains filtered or unexported fields
}

对rtp包进行解封装，并修复时间戳，包括时间戳跳跃

func (*RTPDemuxer) Push added in v3.3.12 

func (r *RTPDemuxer) Push(rtpRaw []byte)

type RTPNalu added in v3.2.3 

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

type RTPVideo 

type RTPVideo struct {
	RTPDemuxer `json:"-"`
	*VideoTrack
	// contains filtered or unexported fields
}

type RingBuffer added in v3.1.0 

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

func (RingBuffer) Clone added in v3.1.0 

func (rb RingBuffer) Clone() *RingBuffer

func (*RingBuffer) Current added in v3.1.0 

func (rb *RingBuffer) Current() *LockItem

func (*RingBuffer) CurrentValue added in v3.1.0 

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

func (*RingBuffer) Dispose added in v3.2.5 

func (rb *RingBuffer) Dispose()

func (*RingBuffer) GetNext added in v3.1.0 

func (rb *RingBuffer) GetNext() *LockItem

func (*RingBuffer) Init added in v3.1.0 

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

func (*RingBuffer) MoveNext added in v3.1.0 

func (rb *RingBuffer) MoveNext()

func (*RingBuffer) NextValue added in v3.1.0 

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

func (*RingBuffer) Read added in v3.1.0 

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

func (*RingBuffer) ReadLoop added in v3.1.7 

func (rb *RingBuffer) ReadLoop(handler interface{}, async bool)

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

func (*RingBuffer) ReadLoopConditional added in v3.1.8 

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

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

func (*RingBuffer) ReadLoopConditionalGo added in v3.3.14 

func (r *RingBuffer) ReadLoopConditionalGo(handler interface{}, goon func() bool)

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

func (*RingBuffer) Step added in v3.3.1 

func (rb *RingBuffer) Step()

func (RingBuffer) SubRing added in v3.1.0 

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

func (*RingBuffer) Write added in v3.1.0 

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

type Stream 

type Stream struct {
	URL            string //远程地址，仅远程拉流有值
	StreamContext  `json:"-"`
	StreamPath     string
	Type           string        //流类型，来自发布者
	StartTime      time.Time     //流的创建时间
	Subscribers    []*Subscriber // 订阅者
	VideoTracks    Tracks
	AudioTracks    Tracks
	DataTracks     Tracks
	AutoCloseAfter *int              //当无人订阅时延迟N秒后自动停止发布
	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 added in v3.2.0 

func Publish(streamPath, t string) *Stream

Publish 直接发布

func (*Stream) Close 

func (r *Stream) Close()

func (*Stream) NewAudioTrack 

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

func (*Stream) NewDataTrack added in v3.3.1 

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(codecID 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) WaitAudioTrack 

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

TODO: 触发转码逻辑

func (*Stream) WaitDataTrack added in v3.3.1 

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 added in v3.2.4 

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

func (*StreamCollection) ToList 

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

type StreamContext added in v3.3.7 

type StreamContext struct {
	context.Context

	IsTimeout bool
	// contains filtered or unexported fields
}

func (*StreamContext) Err added in v3.3.7 

func (r *StreamContext) Err() error

func (*StreamContext) Update added in v3.3.7 

func (r *StreamContext) Update()

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 added in v3.1.8 

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 added in v3.3.1 

func (v *VideoPack) ResetNALUs()

func (*VideoPack) SetNalu0 added in v3.3.1 

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 added in v3.1.3 

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

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.