engine

package module
v4.10.2 Latest Latest
Warning

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

 Go to latest Published: Jan 19, 2023 License: MIT Imports: 37 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:""}
  • 停止推流 /api/stoppush?url=xxx 停止向xxx推流 ，成功返回ok

引擎默认配置

global:
  loglevel: info # 日志级别，可选值：debug,info,warn,error,panic,fatal
  http:
    listenaddr: :8080 # 网关地址，用于访问API
    listenaddrtls: ""  # 用于HTTPS方式访问API的端口配置
    certfile: ""
    keyfile: ""
    cors: true  # 是否自动添加cors头
    username: ""  # 用户名和密码，用于API访问时的基本身份认证
    password: ""
    readtimeout: 0 # 读取超时时间，单位秒，0为不限制
    writetimeout: 0 # 写入超时时间，单位秒，0为不限制
    idletimeout: 0 # 空闲超时时间，单位秒，0为不限制
  publish:
      pubaudio: true # 是否发布音频流
      pubvideo: true # 是否发布视频流
      kickexist: false # 剔出已经存在的发布者，用于顶替原有发布者
      publishtimeout: 10 # 发布流默认过期时间单位秒，超过该时间发布者没有恢复流将被删除
      delayclosetimeout: 0 # 自动关闭触发后延迟的秒数(期间内如果有新的订阅则取消触发关闭)，0为关闭该功能，保持连接。
      waitclosetimeout: 0 # 发布者断开后等待时间，超过该时间发布者没有恢复流将被删除，0为关闭该功能，由订阅者决定是否删除
  subscribe:
      subaudio: true # 是否订阅音频流
      subvideo: true # 是否订阅视频流
      iframeonly: false # 只订阅关键帧
      waittimeout: 10 # 等待发布者的秒数，用于订阅尚未发布的流
  rtpreorder : true # 启用RTP包乱序重排
  enableavcc : true  # 启用AVCC格式缓存，用于rtmp协议
  enablertp : true # 启用rtp格式缓存，用于rtsp、websocket、gb28181协议
  enableauth: true # 启用鉴权,详细查看鉴权机制
  enablesubevent: true # 启用订阅事件，用于订阅者上下线事件,关闭可以提高性能
  rtpreoderbufferlen: 50 # rtp乱序重排缓存长度
  speedlimit: 0 # 限速超时时间（毫秒）0为不限速，对于读取文件这类流需要限速，否则读取过快
  eventbussize: 10 # 事件总线缓存大小，事件较多时容易堵阻塞线程，需要增大缓存
  console: 
    server : console.monibuca.com:4242 # 连接远程控制台的地址
    secret: "" # 远程控制台的秘钥
    publicaddr: "" # 实例公网地址，提供远程控制台访问的地址，不配置的话使用自动识别的地址
    publicaddrtls: "" # 实例公网地址，提供远程控制台访问的地址，不配置的话使用自动识别的地址（https）

配置覆盖机制

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

流的状态图

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

鉴权机制

单独鉴权

引擎中定义如下两个接口，插件中的发布者或者订阅者可以实现这两个接口，引擎会在发布或者订阅时调用这两个接口进行鉴权

type AuthSub interface {
	OnAuth(*util.Promise[ISubscriber]) error
}

type AuthPub interface {
	OnAuth(*util.Promise[IPublisher]) error
}
  • OnAuth返回错误即鉴权失败
  • Promise方便异步鉴权，可以后续调用其Resolve或Reject方法进行鉴权结果的返回

全局鉴权

引擎中定义如下两个全局函数的变量，插件中可以对这两个变量进行赋值，引擎会在发布或者订阅时调用这两个接口进行鉴权

var OnAuthSub func(p *util.Promise[ISubscriber]) error
var OnAuthPub func(p *util.Promise[IPublisher]) error

** 注意：如果单独鉴权和全局鉴权同时存在，优先使用单独鉴权 ** ** 全局鉴权函数可以被多次覆盖，所以需要自己实现鉴权逻辑的合并 **

Http中间件

在HTTPConfig接口中增加了AddMiddleware方法，可以通过该方法添加中间件，中间件的定义如下

type Middleware func(string, http.Handler) http.Handler
type HTTPConfig interface {
	GetHTTPConfig() *HTTP
	Listen(ctx context.Context) error
	Handle(string, http.Handler)
	AddMiddleware(Middleware)
}

中间件的添加必须在FirstConfig之前，也就是在Listen之前 例如：

type MyMiddlewareConfig struct {
  	config.HTTP
}
var myMiddlewareConfig = &MyMiddlewareConfig{}
func init(){
  myMiddlewareConfig.AddMiddleware(func(pattern string, handler http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
      // do something
      handler.ServeHTTP(w, r)
    })
  })
}

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
)
View Source 
const (
	SUBSTATE_INIT = iota
	SUBSTATE_FIRST
	SUBSTATE_NORMAL
)

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     chan any
)
View Source 
var ActionNames = [...]string{"publish", "timeout", "publish lost", "close", "last leave", "first enter", "no tracks"}
View Source 
var ErrBadName = errors.New("Stream Already Exist")
View Source 
var ErrNoPullConfig = errors.New("no pull config")
View Source 
var ErrNoPushConfig = errors.New("no push config")
View Source 
var ErrPublisherLost = errors.New("Publisher Lost")
View Source 
var ErrStreamIsClosed = errors.New("Stream Is Closed")
View Source 
var ErrStreamNotExist = errors.New("stream not exist")
View Source 
var OnAuthPub func(p *util.Promise[IPublisher]) error
View Source 
var OnAuthSub func(p *util.Promise[ISubscriber]) error
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 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[[]byte]

type AudioFrame 

type AudioFrame AVFrame[[]byte]

type AudioRTP added in v4.4.0 

type AudioRTP RTPFrame

type AuthPub added in v4.9.0 

type AuthPub interface {
	OnAuth(*util.Promise[IPublisher]) error
}

type AuthSub added in v4.9.0 

type AuthSub interface {
	OnAuth(*util.Promise[ISubscriber]) error
}

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 DefaultYaml added in v4.9.6 

type DefaultYaml string

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_replay_rtpdump added in v4.8.6 

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

func (*GlobalConfig) API_replay_ts added in v4.9.8 

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

func (*GlobalConfig) API_stopPush added in v4.7.2 

func (conf *GlobalConfig) API_stopPush(w 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 HasAnnexB added in v4.7.0 

type HasAnnexB interface {
	GetAnnexB() (r net.Buffers)
}

type IIO 

type IIO interface {
	IsClosed() bool
	OnEvent(any)
	Stop()
	SetIO(any)
	SetParentCtx(context.Context)
	SetStuff(...any)
	IsShutdown() bool
	// contains filtered or unexported methods
}

type IO 

type IO 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
	Spesific           IIO `json:"-"`
}

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

func (*IO) GetIO 

func (io *IO) GetIO() *IO

func (*IO) IsClosed 

func (io *IO) IsClosed() bool

func (*IO) IsShutdown added in v4.9.0 

func (io *IO) IsShutdown() bool

func (*IO) OnEvent 

func (i *IO) OnEvent(event any)

func (*IO) SetIO 

func (i *IO) SetIO(conn any)

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

func (*IO) SetParentCtx 

func (i *IO) SetParentCtx(parent context.Context)

SetParentCtx（可选）

func (*IO) SetStuff added in v4.8.5 

func (i *IO) SetStuff(stuffs ...any)

SetStuff（可选） 设置Writer、Reader、Closer、Context和本IO关联

func (*IO) Stop 

func (io *IO) Stop()

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

type IOConfig 

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

type IPublisher 

type IPublisher interface {
	IIO
	GetPublisher() *Publisher
	// contains filtered or unexported methods
}

type IPuller 

type IPuller interface {
	IPublisher
	Connect() error
	Pull() error
	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
	GetSubscriber() *Subscriber
	IsPlaying() bool
	PlayRaw()
	PlayBlock(byte)
	PlayFLV()
	Stop()
}

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]
	GetDecoderConfiguration() DecoderConfiguration[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        //插件版本
	Yaml               string        //配置文件中的配置项

	RawConfig   config.Config //最终合并后的配置的map形式方便查询
	Modified    config.Config //修改过的配置项
	*zap.Logger `json:"-"`
	// contains filtered or unexported fields
}

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 int) (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

Subscribe 订阅一个流，如果流不存在则创建一个等待流

func (*Plugin) SubscribeBlock 

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

SubscribeBlock 阻塞订阅一个流，直到订阅结束

func (*Plugin) SubscribeExist added in v4.8.8 

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

SubscribeExist 订阅已经存在的流

func (*Plugin) Update 

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

Update 热更新配置

type Publisher 

type Publisher struct {
	IO
	Config            *config.Publish
	common.AudioTrack `json:"-"`
	common.VideoTrack `json:"-"`
}

func (*Publisher) Equal 

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

func (*Publisher) GetPublisher added in v4.9.0 

func (p *Publisher) GetPublisher() *Publisher

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 RTPDumpPublisher added in v4.8.6 

type RTPDumpPublisher struct {
	Publisher
	DumpFile string
	VCodec   codec.VideoCodecID
	ACodec   codec.AudioCodecID
	// contains filtered or unexported fields
}

func (*RTPDumpPublisher) OnEvent added in v4.8.6 

func (t *RTPDumpPublisher) OnEvent(event any)

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 `json:"-"`
}

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
	StateEvent  StateEvent  // 进入当前状态的事件
	Subscribers Subscribers // 订阅者
	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 (r *Stream) Close()

func (*Stream) IsClosed 

func (r *Stream) IsClosed() bool

func (*Stream) IsShutdown added in v4.6.5 

func (r *Stream) IsShutdown() 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 StreamList added in v4.8.2 

type StreamList []*Stream

func GetSortedStreamList added in v4.8.2 

func GetSortedStreamList() StreamList

func (StreamList) Len added in v4.8.2 

func (l StreamList) Len() int

func (StreamList) Less added in v4.8.2 

func (l StreamList) Less(i, j int) bool

func (StreamList) Sort added in v4.8.2 

func (l StreamList) Sort()

func (StreamList) Swap added in v4.8.2 

func (l StreamList) Swap(i, j int)

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
	IsInternal  bool //是否内部订阅,不放入订阅列表
	Config      *config.Subscribe
	TrackPlayer `json:"-"`
}

Subscriber 订阅者实体定义

func (*Subscriber) AddTrack 

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

func (*Subscriber) GetSubscriber added in v4.9.0 

func (s *Subscriber) GetSubscriber() *Subscriber

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 Subscribers added in v4.9.3 

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

func (*Subscribers) AbortWait added in v4.9.5 

func (s *Subscribers) AbortWait()

func (*Subscribers) Add added in v4.9.3 

func (s *Subscribers) Add(suber ISubscriber, wait *waitTracks)

func (*Subscribers) Broadcast added in v4.9.3 

func (s *Subscribers) Broadcast(event any)

func (*Subscribers) Delete added in v4.9.3 

func (s *Subscribers) Delete(suber ISubscriber)

func (*Subscribers) Dispose added in v4.9.5 

func (s *Subscribers) Dispose()

func (*Subscribers) Init added in v4.9.5 

func (s *Subscribers) Init()

func (*Subscribers) Len added in v4.9.5 

func (s *Subscribers) Len() int

func (*Subscribers) MarshalJSON added in v4.9.3 

func (s *Subscribers) MarshalJSON() ([]byte, error)

func (*Subscribers) OnPublisherLost added in v4.9.5 

func (s *Subscribers) OnPublisherLost(event StateEvent)

func (*Subscribers) OnTrack added in v4.9.5 

func (s *Subscribers) OnTrack(track common.Track)

func (*Subscribers) Pick added in v4.9.3 

func (s *Subscribers) Pick() ISubscriber

func (*Subscribers) RangeAll added in v4.9.5 

func (s *Subscribers) RangeAll(f func(sub ISubscriber, wait *waitTracks))

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) OnEvent 

func (t *TSPublisher) OnEvent(event any)

func (*TSPublisher) OnPmtStream 

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

func (*TSPublisher) ReadPES added in v4.9.8 

func (t *TSPublisher) ReadPES()

type TrackPlayer 

type TrackPlayer struct {
	context.Context    `json:"-"`
	context.CancelFunc `json:"-"`
	Audio              PlayContext[*track.Audio, []byte]
	Video              PlayContext[*track.Video, NALUSlice]
	SkipTS             uint32 //跳过的时间戳
	FirstAbsTS         uint32 //订阅起始时间戳
}

func (*TrackPlayer) ReadAudio added in v4.6.9 

func (tp *TrackPlayer) ReadAudio() (ap *AVFrame[[]byte])

func (*TrackPlayer) ReadVideo added in v4.6.9 

func (tp *TrackPlayer) ReadVideo() (vp *AVFrame[NALUSlice])

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 UnsubscribeEvent added in v4.9.0 

type UnsubscribeEvent struct {
	Subscriber ISubscriber
}

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)
func copyBuffers(b net.Buffers) (r net.Buffers) {
	return append(r, b...)
}

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.