engine

package module
v4.6.4 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2022 License: MIT Imports: 35 Imported by: 56

README

m7s v4核心引擎

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

引擎的基本功能

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

引擎自带HTTP接口

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

引擎默认配置

global:
  loglevel: info # 日志级别，可选值：debug,info,warn,error,panic,fatal
  http:
    # 网关地址，用于访问API
    listenaddr: :8080
    # 用于HTTPS方式访问API的端口配置
    listenaddrtls: ""
    certfile: ""
    keyfile: ""
    # 是否自动添加cors头
    cors: true
    # 用户名和密码，用于API访问时的基本身份认证
    username: ""
    password: ""
  publish:
      # 是否发布音频流
      pubaudio: true
      # 是否发布视频流
      pubvideo: true
      # 剔出已经存在的发布者，用于顶替原有发布者
      kickexist: false
      # 发布流默认过期时间单位秒，超过该时间发布者没有恢复流将被删除
      publishtimeout: 10
      # 自动关闭触发后延迟的秒数(期间内如果有新的订阅则取消触发关闭)，0为关闭该功能，保持连接。
      delayclosetimeout: 0
      # 发布者断开后等待时间，超过该时间发布者没有恢复流将被删除，0为关闭该功能，由订阅者决定是否删除
      waitclosetimeout: 0
  subscribe:
      # 是否订阅音频流
      subaudio: true
      # 是否订阅视频流
      subvideo: true
      # 只订阅关键帧
      iframeonly: false
      # 等待发布者的秒数，用于订阅尚未发布的流
      waittimeout: 10
  # 启用RTP包乱序重排
  rtpreorder : false
  # 启用AVCC格式缓存，用于rtmp协议
  enableavcc : true
  # 启用rtp格式缓存，用于rtsp、websocket、gb28181协议
  enablertp : true
  console: 
    # 连接远程控制台的地址
    server : wss://console.monibuca.com/ws/v1
    # 远程控制台的秘钥
    secret: ""
    # 实例公网地址，提供远程控制台访问的地址，不配置的话使用自动识别的地址
    publicaddr: ""
    # 实例公网地址，提供远程控制台访问的地址，不配置的话使用自动识别的地址（https）
    publicaddrtls: ""

配置覆盖机制

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

流的状态图

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

Documentation

Index

Constants

View Source 
const (
	NO_SUCH_CONIFG = "no such config"
	NO_SUCH_STREAM = "no such stream"
)
View Source 
const (
	SUBTYPE_RAW = iota
	SUBTYPE_AVCC
	SUBTYPE_RTP
	SUBTYPE_FLV
)

Variables

View Source 
var (
	SysInfo struct {
		StartTime time.Time //启动时间
		LocalIP   string
		Version   string
	}
	ExecPath = os.Args[0]
	ExecDir  = filepath.Dir(ExecPath)
	// ConfigRaw 配置信息的原始数据
	ConfigRaw    []byte
	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 Pullers sync.Map
View Source 
var Pushers sync.Map
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]

type AudioFrame 

type AudioFrame AVFrame[AudioSlice]

type AudioRTP added in v4.4.0 

type AudioRTP 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 FLVFrame added in v4.3.0 

type FLVFrame net.Buffers

func (FLVFrame) WriteTo added in v4.3.0 

func (f FLVFrame) WriteTo(w io.Writer) (int64, error)

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_list_pull added in v4.6.4 

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

func (*GlobalConfig) API_list_push added in v4.6.4 

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

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 IIO 

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

type IO 

type IO[C IOConfig] 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             IIO `json:"-"`
}

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

func (*IO[C]) GetConfig 

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

func (*IO[C]) GetIO 

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

func (*IO[C]) IsClosed 

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

func (*IO[C]) OnEvent 

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

func (*IO[C]) SetIO 

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

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

func (*IO[C]) SetParentCtx 

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

SetParentCtx（可选）

func (*IO[C]) Stop 

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

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

type IOConfig 

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

type IPublisher 

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

	GetIO() *IO[config.Publish]
	// 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]
	GetConfig() *config.Subscribe
	IsPlaying() bool
	PlayRaw()
	PlayBlock(byte)
	PlayFLV()
	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

	Frame *AVFrame[R]
	// 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, t byte) (err error)

func (*Plugin) Update 

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

Update 热更新配置

type Publisher 

type Publisher struct {
	IO[config.Publish]
	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      Tracks
	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   time.Time
	Type        string
	BPS         int
}

type StreamTimeoutConfig 

type StreamTimeoutConfig struct {
	PublishTimeout    time.Duration //发布者无数据后超时
	DelayCloseTimeout time.Duration //发布者丢失后等待
}

type Subscriber 

type Subscriber struct {
	IO[config.Subscribe]
	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(subType byte)

PlayBlock 阻塞式读取数据

func (*Subscriber) PlayFLV added in v4.3.0 

func (s *Subscriber) PlayFLV()

func (*Subscriber) PlayRTP added in v4.4.0 

func (s *Subscriber) PlayRTP()

func (*Subscriber) PlayRaw added in v4.3.0 

func (s *Subscriber) PlayRaw()

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]
	SkipTS             uint32 //跳过的时间戳
	FirstAbsTS         uint32 //订阅起始时间戳
}

type TrackRemoved 

type TrackRemoved struct {
	Track
}

type Tracks added in v4.5.1 

type Tracks struct {
	util.Map[string, Track]
}

func (*Tracks) MarshalJSON added in v4.5.1 

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

type VideoDeConf 

type VideoDeConf DecoderConfiguration[NALUSlice]

func (VideoDeConf) GetAnnexB added in v4.2.0 

func (v VideoDeConf) GetAnnexB() (r net.Buffers)

type VideoFrame 

type VideoFrame AVFrame[NALUSlice]

func (*VideoFrame) GetAnnexB added in v4.2.0 

func (v *VideoFrame) GetAnnexB() (r net.Buffers)

type VideoRTP added in v4.4.0 

type VideoRTP RTPFrame

Source Files

View all Source files

Directories

Path Synopsis
mpegts
bits
bits/bufio
bits/pio

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.