engine

package module
v4.1.4 Latest Latest
Warning

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

 Go to latest Published: Jun 7, 2022 License: MIT Imports: 34 Imported by: 56

README

m7s v4核心引擎

该项目为m7s的引擎部分，该部分逻辑是流媒体服务器的核心转发逻辑。包含了一个插件的引入机制，其他功能均由插件实现

引擎的基本功能

  • 引擎初始化会加载配置文件，引入的插件会自动注册到引擎中
  • 配置文件中配置会被解析并覆盖插件的默认配置
  • 引擎提供配置热更新机制（具体热更新逻辑由插件实现）
  • 读取插件的特殊方法，将其注册为可供HTTP访问的API接口
  • 具有发布功能的插件，可以将流注入到引擎中
  • 具有订阅功能的插件，可以从引擎中订阅到流
  • 引擎会将流中的数据放入RingBuffer中缓存，以便插件可以获取数据
  • 引擎提供了从远端拉流和以及向远端推流的基础框架
  • 引擎包了zap日志框架
  • 引擎提供事件总线机制，可以对所有插件广播事件

引擎自带HTTP接口

  • 终止某一个流 /api/closeStream?streamPath=xxx
  • 获取engine信息 /api/sysInfo 返回值{Version:xxx,StartTime:xxx}
  • 获取系统基本情况 /api/summary 返回值Summary数据
  • 获取所有插件信息 /api/plugins 返回值Plugin数据
  • 获取指定的配置信息 /api/getconfig?name=xxx 返回xxx插件的配置信息，如果不带参数或参数为空则返回全局配置
  • 修改并保存配置信息 /api/modifyconfig?name=xxx 修改xxx插件的配置信息,在请求的body中传入修改后的配置json字符串
  • 热更新配置信息 /api/updateconfig?name=xxx 热更新xxx插件的配置信息，如果不带参数或参数为空则热更新全局配置

引擎默认配置

global:
  http:
    # 网关地址，用于访问API
    listenaddr: :8080
    # 用于HTTPS方式访问API的端口配置
    listenaddrtls: ""
    certfile: ""
    keyfile: ""
    # 是否自动添加cors头
    cors: true
    # 用户名和密码，用于API访问时的基本身份认证
    username: ""
    password: ""
  publish:
      # 是否发布音频流
      pubaudio: true
      # 是否发布视频流
      pubvideo: true
      # 剔出已经存在的发布者，用于顶替原有发布者
      kickexist: false
      # 发布流默认过期时间单位秒，超过该时间发布者没有恢复流将被删除
      publishtimeout: 10
      # 自动关闭触发后延迟的秒数(期间内如果有新的订阅则取消触发关闭)
      waitclosetimeout: 0
  subscribe:
      # 是否订阅音频流
      subaudio: true
      # 是否订阅视频流
      subvideo: true
      # 只订阅关键帧
      iframeonly: false
      # 等待发布者的秒数，用于订阅尚未发布的流
      waittimeout: 10
  # 启用RTP包乱序重排
  rtpreorder : false
  # 启用AVCC格式缓存，用于rtmp协议
  enableavcc : true
  # 启用rtp格式缓存，用于rtsp、websocket、gb28181协议
  enablertp : true
  # 启用flv格式缓存，用于HDL协议，以及flv格式写文件
  enableflv : true
  # 连接远程控制台的地址
  consoleurl : wss://console.monibuca.com/ws/v1
  # 远程控制台的秘钥
  secret: ""

配置覆盖机制

  • 如果不存在配置文件，将使用默认配置，该配置值为代码中写死的配置值
  • 如果存在配置文件，则使用配置文件中的值覆盖默认值
  • http、publish、subscribe三个配置遵循优先级顺序
  1. 如果发布流或者订阅流中包含对应的参数，则优先使用
  2. 其次，查找对应插件的配置项中是否包含配置项
  3. 最后，使用全局配置中的配置

流的状态图

stateDiagram-v2
    [*] --> ⌛等待发布者 : 创建
    ⌛等待发布者 --> 🟢正在发布 :发布
    ⌛等待发布者 --> 🔴已关闭 :关闭
    ⌛等待发布者 --> 🔴已关闭  :超时
    ⌛等待发布者 --> 🔴已关闭  :最后订阅者离开
    🟢正在发布 --> ⌛等待发布者: 发布者断开
    🟢正在发布 --> 🟡等待关闭: 最后订阅者离开
    🟢正在发布 --> 🔴已关闭  :关闭
    🟡等待关闭 --> 🟢正在发布 :第一个订阅者进入
    🟡等待关闭 --> 🔴已关闭  :关闭
    🟡等待关闭 --> 🔴已关闭  :超时
    🟡等待关闭 --> 🔴已关闭  :发布者断开

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ExecPath = os.Args[0]
	ExecDir  = filepath.Dir(ExecPath)
	// ConfigRaw 配置信息的原始数据
	ConfigRaw    []byte
	StartTime    time.Time                  //启动时间
	Plugins      = make(map[string]*Plugin) // Plugins 所有的插件配置
	EngineConfig = &GlobalConfig{
		Engine: config.Global,
	}

	Engine       = InstallPlugin(EngineConfig)              //复用安装插件逻辑，将全局配置信息注入，并启动server
	MergeConfigs = []string{"Publish", "Subscribe", "HTTP"} //需要合并配置的属性项，插件若没有配置则使用全局配置
	EventBus     = make(chan any, 10)
)
View Source 
var ActionNames = [...]string{"publish", "timeout", "publish lost", "close", "last leave", "first enter", "no tracks"}
View Source 
var BadNameErr = errors.New("Bad Name")
View Source 
var NoPullConfigErr = errors.New("no pull config")
View Source 
var NoPushConfigErr = errors.New("no push config")
View Source 
var StateNames = [...]string{"⌛", "🟢", "🟡", "🔴"}
View Source 
var StreamFSM = [len(StateNames)]map[StreamAction]StreamState{
	{
		ACTION_PUBLISH:   STATE_PUBLISHING,
		ACTION_TIMEOUT:   STATE_CLOSED,
		ACTION_LASTLEAVE: STATE_CLOSED,
		ACTION_CLOSE:     STATE_CLOSED,
	},
	{
		ACTION_PUBLISHLOST: STATE_WAITPUBLISH,
		ACTION_LASTLEAVE:   STATE_WAITCLOSE,
		ACTION_CLOSE:       STATE_CLOSED,
	},
	{
		ACTION_PUBLISHLOST: STATE_CLOSED,
		ACTION_TIMEOUT:     STATE_CLOSED,
		ACTION_FIRSTENTER:  STATE_PUBLISHING,
		ACTION_CLOSE:       STATE_CLOSED,
	},
	{},
}
View Source 
var StreamIsClosedErr = errors.New("Stream Is Closed")
View Source 
var Streams = util.Map[string, *Stream]{Map: make(map[string]*Stream)}

Streams 所有的流集合

Functions

func Run 

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

Run 启动Monibuca引擎，传入总的Context，可用于关闭所有

Types

type AudioDeConf 

type AudioDeConf DecoderConfiguration[AudioSlice]

func (AudioDeConf) GetAVCC 

func (a AudioDeConf) GetAVCC() net.Buffers

func (AudioDeConf) GetFLV 

func (a AudioDeConf) GetFLV() net.Buffers

type AudioFrame 

type AudioFrame AVFrame[AudioSlice]

func (*AudioFrame) GetAVCC 

func (a *AudioFrame) GetAVCC() net.Buffers

func (*AudioFrame) GetFLV 

func (a *AudioFrame) GetFLV() net.Buffers

func (*AudioFrame) GetRTP 

func (a *AudioFrame) GetRTP() []*RTPFrame

type ClientConfig 

type ClientConfig interface {
	config.Pull | config.Push
}

type ClientIO added in v4.1.0 

type ClientIO[C ClientConfig] struct {
	Config         *C
	StreamPath     string // 本地流标识
	RemoteURL      string // 远程服务器地址（用于推拉）
	ReConnectCount int    //重连次数
}

ClientIO 作为Client角色(Puller，Pusher)的公共结构体

type FirstConfig 

type FirstConfig config.Config

type GlobalConfig 

type GlobalConfig struct {
	*config.Engine
}

func (*GlobalConfig) API_closeStream 

func (conf *GlobalConfig) API_closeStream(w http.ResponseWriter, r *http.Request)

func (*GlobalConfig) API_getConfig 

func (conf *GlobalConfig) API_getConfig(w http.ResponseWriter, r *http.Request)

API_getConfig 获取指定的配置信息

func (*GlobalConfig) API_modifyConfig 

func (conf *GlobalConfig) API_modifyConfig(w http.ResponseWriter, r *http.Request)

API_modifyConfig 修改并保存配置

func (*GlobalConfig) API_plugins added in v4.1.4 

func (conf *GlobalConfig) API_plugins(rw http.ResponseWriter, r *http.Request)

func (*GlobalConfig) API_stream added in v4.1.0 

func (conf *GlobalConfig) API_stream(rw http.ResponseWriter, r *http.Request)

func (*GlobalConfig) API_summary 

func (conf *GlobalConfig) API_summary(rw http.ResponseWriter, r *http.Request)

func (*GlobalConfig) API_sysInfo 

func (conf *GlobalConfig) API_sysInfo(rw http.ResponseWriter, r *http.Request)

func (*GlobalConfig) API_updateConfig 

func (conf *GlobalConfig) API_updateConfig(w http.ResponseWriter, r *http.Request)

API_updateConfig 热更新配置

func (*GlobalConfig) ServeHTTP 

func (conf *GlobalConfig) ServeHTTP(rw http.ResponseWriter, r *http.Request)

type HaveAVCC 

type HaveAVCC interface {
	GetAVCC() net.Buffers
}

type HaveFLV 

type HaveFLV interface {
	GetFLV() net.Buffers
}

type HaveRTP 

type HaveRTP interface {
	GetRTP() []*RTPFrame
}

type IIO 

type IIO interface {
	IsClosed() bool
	OnEvent(any)
	Stop()
	SetIO(any)
	SetParentCtx(context.Context)
}

type IO 

type IO[C IOConfig, S IIO] struct {
	ID                 string
	Type               string
	context.Context    `json:"-"` //不要直接设置，应当通过OnEvent传入父级Context
	context.CancelFunc `json:"-"` //流关闭是关闭发布者或者订阅者
	*zap.Logger        `json:"-"`
	StartTime          time.Time //创建时间
	Stream             *Stream   `json:"-"`
	io.Reader          `json:"-"`
	io.Writer          `json:"-"`
	io.Closer          `json:"-"`
	Args               url.Values
	Config             *C `json:"-"`
	Spesic             S  `json:"-"`
}

发布者或者订阅者的共用结构体

func (*IO[C, S]) GetConfig 

func (io *IO[C, S]) GetConfig() *C

func (*IO[C, S]) GetIO 

func (io *IO[C, S]) GetIO() *IO[C, S]

func (*IO[C, S]) IsClosed 

func (io *IO[C, S]) IsClosed() bool

func (*IO[C, S]) OnEvent 

func (i *IO[C, S]) OnEvent(event any)

func (*IO[C, S]) SetIO 

func (i *IO[C, S]) SetIO(conn any)

SetIO（可选） 设置Writer、Reader、Closer

func (*IO[C, S]) SetParentCtx 

func (i *IO[C, S]) SetParentCtx(parent context.Context)

SetParentCtx（可选）

func (*IO[C, S]) Stop 

func (io *IO[C, S]) Stop()

Stop 停止订阅或者发布，由订阅者或者发布者调用

type IOConfig 

type IOConfig interface {
	config.Publish | config.Subscribe
}

type IPublisher 

type IPublisher interface {
	IIO
	GetConfig() *config.Publish

	GetIO() *IO[config.Publish, IPublisher]
	// contains filtered or unexported methods
}

type IPuller 

type IPuller interface {
	IPublisher
	Connect() error
	Pull()
	Reconnect() bool
	// contains filtered or unexported methods
}

type IPusher 

type IPusher interface {
	ISubscriber
	Push() error
	Connect() error

	Reconnect() bool
	// contains filtered or unexported methods
}

type ISubscriber 

type ISubscriber interface {
	IIO

	GetIO() *IO[config.Subscribe, ISubscriber]
	GetConfig() *config.Subscribe
	IsPlaying() bool
	PlayBlock()
	Stop()
	// contains filtered or unexported methods
}

type NetWorkInfo 

type NetWorkInfo struct {
	Name         string
	Receive      uint64
	Sent         uint64
	ReceiveSpeed uint64
	SentSpeed    uint64
}

NetWorkInfo 网速信息

type PlayContext 

type PlayContext[T interface {
	GetDecConfSeq() int
	ReadRing() *AVRing[R]
}, R RawSlice] struct {
	Track T

	First AVFrame[R] `json:"-"`
	// contains filtered or unexported fields
}

func (*PlayContext[T, R]) MarshalJSON 

func (p *PlayContext[T, R]) MarshalJSON() ([]byte, error)

type Plugin 

type Plugin struct {
	context.Context    `json:"-"`
	context.CancelFunc `json:"-"`
	Name               string        //插件名称
	Config             config.Plugin `json:"-"` //插件配置
	Version            string        //插件版本
	RawConfig          config.Config //配置的map形式方便查询
	Modified           config.Config //修改过的配置项
	*zap.Logger        `json:"-"`
}

Plugin 插件信息

func InstallPlugin 

func InstallPlugin(config config.Plugin) *Plugin

InstallPlugin 安装插件，传入插件配置生成插件信息对象

func (*Plugin) Publish 

func (opt *Plugin) Publish(streamPath string, pub IPublisher) error

func (*Plugin) Pull 

func (opt *Plugin) Pull(streamPath string, url string, puller IPuller, save bool) (err error)

func (*Plugin) Push 

func (opt *Plugin) Push(streamPath string, url string, pusher IPusher, save bool) (err error)

func (*Plugin) Save 

func (opt *Plugin) Save() error

func (*Plugin) Subscribe 

func (opt *Plugin) Subscribe(streamPath string, sub ISubscriber) error

func (*Plugin) SubscribeBlock 

func (opt *Plugin) SubscribeBlock(streamPath string, sub ISubscriber) (err error)

func (*Plugin) Update 

func (opt *Plugin) Update(conf config.Config)

Update 热更新配置

type Publisher 

type Publisher struct {
	IO[config.Publish, IPublisher]
	common.AudioTrack `json:"-"`
	common.VideoTrack `json:"-"`
}

func (*Publisher) Equal 

func (p *Publisher) Equal(p2 IPublisher) bool

func (*Publisher) OnEvent 

func (p *Publisher) OnEvent(event any)

func (*Publisher) Stop 

func (p *Publisher) Stop()

func (*Publisher) WriteAVCCAudio added in v4.1.0 

func (p *Publisher) WriteAVCCAudio(ts uint32, frame common.AVCCFrame)

func (*Publisher) WriteAVCCVideo added in v4.1.0 

func (p *Publisher) WriteAVCCVideo(ts uint32, frame common.AVCCFrame)

type Puller 

type Puller struct {
	ClientIO[config.Pull]
}

用于远程拉流的发布者

func (*Puller) Reconnect 

func (pub *Puller) Reconnect() (ok bool)

是否需要重连

type Pusher 

type Pusher struct {
	ClientIO[config.Push]
}

func (*Pusher) Reconnect 

func (pub *Pusher) Reconnect() bool

是否需要重连

type SEKick 

type SEKick struct {
}

type SEclose 

type SEclose struct {
	StateEvent
}

type SEpublish 

type SEpublish struct {
	StateEvent
}

type SEwaitClose 

type SEwaitClose struct {
	StateEvent
}

type SEwaitPublish 

type SEwaitPublish struct {
	StateEvent
	Publisher IPublisher
}

type StateEvent 

type StateEvent struct {
	Action StreamAction
	From   StreamState
	Stream *Stream
}

func (StateEvent) Next 

func (se StateEvent) Next() (next StreamState, ok bool)

type Stream 

type Stream struct {
	*zap.Logger
	StartTime time.Time //创建时间
	StreamTimeoutConfig
	Path        string
	Publisher   IPublisher
	State       StreamState
	Subscribers []ISubscriber // 订阅者
	Tracks      map[string]Track
	AppName     string
	StreamName  string
	// contains filtered or unexported fields
}

Stream 流定义

func FilterStreams 

func FilterStreams[T IPublisher]() (ss []*Stream)

func (*Stream) AddTrack 

func (s *Stream) AddTrack(t Track)

func (*Stream) Close 

func (s *Stream) Close()

func (*Stream) IsClosed 

func (r *Stream) IsClosed() bool

func (*Stream) NewDataTrack 

func (r *Stream) NewDataTrack(locker sync.Locker) (dt *track.Data)

func (*Stream) Receive 

func (s *Stream) Receive(event any) bool

func (*Stream) RemoveTrack 

func (s *Stream) RemoveTrack(t Track)

func (*Stream) SSRC 

func (s *Stream) SSRC() uint32

func (*Stream) Summary added in v4.1.0 

func (s *Stream) Summary() (r StreamSummay)

Summary 返回流的简要信息

type StreamAction 

type StreamAction byte
const (
	ACTION_PUBLISH     StreamAction = iota
	ACTION_TIMEOUT                  // 发布流长时间没有数据/长时间没有发布者发布流/等待关闭时间到
	ACTION_PUBLISHLOST              // 发布者意外断开
	ACTION_CLOSE                    // 主动关闭流
	ACTION_LASTLEAVE                // 最后一个订阅者离开
	ACTION_FIRSTENTER               // 第一个订阅者进入
)

type StreamState 

type StreamState byte
const (
	STATE_WAITPUBLISH StreamState = iota // 等待发布者状态
	STATE_PUBLISHING                     // 正在发布流状态
	STATE_WAITCLOSE                      // 等待关闭状态(自动关闭延时开启)
	STATE_CLOSED                         // 流已关闭，不可使用
)

四状态机

type StreamSummay added in v4.1.0 

type StreamSummay struct {
	Path        string
	State       StreamState
	Subscribers int
	Tracks      []string
	StartTime   int64
	Type        string
	BPS         int
}

type StreamTimeoutConfig 

type StreamTimeoutConfig struct {
	WaitTimeout      time.Duration
	PublishTimeout   time.Duration
	WaitCloseTimeout time.Duration
}

type Subscriber 

type Subscriber struct {
	IO[config.Subscribe, ISubscriber]
	TrackPlayer `json:"-"`
}

Subscriber 订阅者实体定义

func (*Subscriber) AddTrack 

func (s *Subscriber) AddTrack(t Track) bool

func (*Subscriber) IsPlaying 

func (s *Subscriber) IsPlaying() bool

func (*Subscriber) OnEvent 

func (s *Subscriber) OnEvent(event any)

func (*Subscriber) PlayBlock 

func (s *Subscriber) PlayBlock()

PlayBlock 阻塞式读取数据

type Summary 

type Summary struct {
	Address string
	Memory  struct {
		Total uint64
		Free  uint64
		Used  uint64
		Usage float64
	}
	CPUUsage float64
	HardDisk struct {
		Total uint64
		Free  uint64
		Used  uint64
		Usage float64
	}
	NetWork []NetWorkInfo
	Streams []StreamSummay
	// contains filtered or unexported fields
}

ServerSummary 系统摘要定义

func (*Summary) Add 

func (s *Summary) Add()

Add 增加订阅者

func (*Summary) Done 

func (s *Summary) Done()

Done 删除订阅者

func (*Summary) Point 

func (s *Summary) Point() *Summary

func (*Summary) Report 

func (s *Summary) Report(slave *Summary)

Report 上报数据

func (*Summary) Running 

func (s *Summary) Running() bool

Running 是否正在采集数据

func (*Summary) Start 

func (s *Summary) Start()

StartSummary 开始定时采集数据，每秒一次

type TSPublisher 

type TSPublisher struct {
	Publisher
	*mpegts.MpegTsStream
	// contains filtered or unexported fields
}

func (*TSPublisher) Feed added in v4.0.1 

func (t *TSPublisher) Feed(r io.Reader) error

func (*TSPublisher) OnEvent 

func (t *TSPublisher) OnEvent(event any)

func (*TSPublisher) OnPES 

func (t *TSPublisher) OnPES(pes mpegts.MpegTsPESPacket)

func (*TSPublisher) OnPmtStream 

func (t *TSPublisher) OnPmtStream(s mpegts.MpegTsPmtStream)

type TrackPlayer 

type TrackPlayer struct {
	context.Context    `json:"-"`
	context.CancelFunc `json:"-"`
	Audio              PlayContext[*track.Audio, AudioSlice]
	Video              PlayContext[*track.Video, NALUSlice]
}

type TrackRemoved 

type TrackRemoved struct {
	Track
}

type VideoDeConf 

type VideoDeConf DecoderConfiguration[NALUSlice]

func (VideoDeConf) GetAVCC 

func (a VideoDeConf) GetAVCC() net.Buffers

func (VideoDeConf) GetFLV 

func (a VideoDeConf) GetFLV() net.Buffers

type VideoFrame 

type VideoFrame AVFrame[NALUSlice]

func (*VideoFrame) GetAVCC 

func (v *VideoFrame) GetAVCC() net.Buffers

func (*VideoFrame) GetFLV 

func (v *VideoFrame) GetFLV() net.Buffers

func (*VideoFrame) GetRTP 

func (v *VideoFrame) GetRTP() []*RTPFrame

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.