manager

Documentation

Index

• Variables
• func AttachVideoHandler(r *router.Router, prefix string, manager *VideoManager, log logging.KVLogger)
• func CORSMiddleware(h fasthttp.RequestHandler) fasthttp.RequestHandler
• func MetricsMiddleware(h fasthttp.RequestHandler) fasthttp.RequestHandler
• func RegisterMetrics()
• func SetLogger(l *zap.SugaredLogger)
• type Gatekeeper
• type HttpAPI
  • func NewHttpAPI(cfg *HttpAPIConfiguration) *HttpAPI
  • func (s HttpAPI) Addr() string
  • func (s HttpAPI) Shutdown() error
  • func (s HttpAPI) Start() error
  • func (s HttpAPI) URL() string
• type HttpAPIConfiguration
  • func ConfigureHttpAPI() *HttpAPIConfiguration
  • func (c *HttpAPIConfiguration) Addr(addr string) *HttpAPIConfiguration
  • func (c *HttpAPIConfiguration) Debug(debug bool) *HttpAPIConfiguration
  • func (c *HttpAPIConfiguration) VideoManager(mgr *VideoManager) *HttpAPIConfiguration
  • func (c *HttpAPIConfiguration) VideoPath(videoPath string) *HttpAPIConfiguration
• type Pool
  • func NewPool() *Pool
  • func (p *Pool) AddQueue(name string, minHits uint, k Gatekeeper)
  • func (p *Pool) Admit(key string, value interface{}) error
  • func (p *Pool) Next() *mfr.Item
  • func (p *Pool) Out() <-chan *mfr.Item
  • func (p *Pool) Start()
  • func (p *Pool) Stop()
• type TranscodingRequest
  • func (r *TranscodingRequest) Complete()
  • func (r *TranscodingRequest) Reject()
  • func (r *TranscodingRequest) Release()
• type VideoLibrary
• type VideoManager
  • func NewManager(lib *library.Library, minHits int) *VideoManager
  • func (m *VideoManager) Library() *library.Library
  • func (m *VideoManager) Pool() *Pool
  • func (m *VideoManager) RequestStatus(sdHash string) int
  • func (m *VideoManager) Requests() <-chan *TranscodingRequest
  • func (m *VideoManager) ResolveStream(uri string) (*TranscodingRequest, error)
  • func (m *VideoManager) Video(uri string) (string, error)

Variables

var (
	QueueLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Name: "transcoding_queue_length",
		Help: "Video queue length",
	}, []string{"queue"})

	QueueHits = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Name: "transcoding_queue_hits",
		Help: "Video queue hits",
	}, []string{"queue"})

	QueueItemAge = prometheus.NewSummaryVec(prometheus.SummaryOpts{
		Name: "transcoding_queue_item_age_seconds",
		Help: "Age of queue items before they get processed",
	}, []string{"queue"})
)

Functions

func AttachVideoHandler

func AttachVideoHandler(r *router.Router, prefix string, manager *VideoManager, log logging.KVLogger)

NewRouter creates a set of HTTP entrypoints that will route requests into video library and video transcoding queue.

func CORSMiddleware

func CORSMiddleware(h fasthttp.RequestHandler) fasthttp.RequestHandler

func MetricsMiddleware

func MetricsMiddleware(h fasthttp.RequestHandler) fasthttp.RequestHandler

func RegisterMetrics

func RegisterMetrics()

func SetLogger

func SetLogger(l *zap.SugaredLogger)

Types

type Gatekeeper

type Gatekeeper func(key string, value interface{}, queue *mfr.Queue) bool

Gatekeeper defines a function that checks if supplied queue item and its value should be admitted to the queue.

type HttpAPI

type HttpAPI struct {
	*HttpAPIConfiguration
	// contains filtered or unexported fields
}

HttpAPI ties HTTP API together and allows to start/shutdown the web server.

func NewHttpAPI

func NewHttpAPI(cfg *HttpAPIConfiguration) *HttpAPI

func (HttpAPI) Addr

func (s HttpAPI) Addr() string

func (HttpAPI) Shutdown

func (s HttpAPI) Shutdown() error

func (HttpAPI) Start

func (s HttpAPI) Start() error

func (HttpAPI) URL

func (s HttpAPI) URL() string

type HttpAPIConfiguration

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

func ConfigureHttpAPI

func ConfigureHttpAPI() *HttpAPIConfiguration

func (*HttpAPIConfiguration) Addr

func (c *HttpAPIConfiguration) Addr(addr string) *HttpAPIConfiguration

func (*HttpAPIConfiguration) Debug

func (c *HttpAPIConfiguration) Debug(debug bool) *HttpAPIConfiguration

func (*HttpAPIConfiguration) VideoManager

func (c *HttpAPIConfiguration) VideoManager(mgr *VideoManager) *HttpAPIConfiguration

func (*HttpAPIConfiguration) VideoPath

func (c *HttpAPIConfiguration) VideoPath(videoPath string) *HttpAPIConfiguration

type Pool

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

Pool contains queues which can admit items based on gatekeeper functions.

func NewPool

func NewPool() *Pool

func (*Pool) AddQueue

func (p *Pool) AddQueue(name string, minHits uint, k Gatekeeper)

AddQueue adds a queue and its gatekeeper function to the pool.

func (*Pool) Admit

func (p *Pool) Admit(key string, value interface{}) error

Admit retries to put item into the first queue that would accept it. Queues are traversed in the same order they are added. If gatekeeper returns an error, admission stops and the error is returned to the caller.

func (*Pool) Next

func (p *Pool) Next() *mfr.Item

Next returns the next item in the queue almost in a non-blocking way.

func (*Pool) Out

func (p *Pool) Out() <-chan *mfr.Item

func (*Pool) Start

func (p *Pool) Start()

Start will launch the cycle of retrieving items out of queues. Should be called after at least one `AddQueue` call. Queues are pooled sequentially.

func (*Pool) Stop

func (p *Pool) Stop()

Stop stops the queue polling routine.

type TranscodingRequest

type TranscodingRequest struct {
	resolve.ResolvedStream
	// contains filtered or unexported fields
}

func (*TranscodingRequest) Complete

func (r *TranscodingRequest) Complete()

func (*TranscodingRequest) Reject

func (r *TranscodingRequest) Reject()

func (*TranscodingRequest) Release

func (r *TranscodingRequest) Release()

type VideoLibrary

type VideoLibrary interface {
	GetURL(sdHash string) (string, error)
	GetAllChannels() ([]db.Channel, error)
}

type VideoManager

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

func NewManager

func NewManager(lib *library.Library, minHits int) *VideoManager

NewManager creates a video library manager with a pool for future transcoding requests.

func (*VideoManager) Library

func (m *VideoManager) Library() *library.Library

func (*VideoManager) Pool

func (m *VideoManager) Pool() *Pool

func (*VideoManager) RequestStatus

func (m *VideoManager) RequestStatus(sdHash string) int

func (*VideoManager) Requests

func (m *VideoManager) Requests() <-chan *TranscodingRequest

Requests returns next transcoding request to be processed. It polls all queues in the pool evenly.

func (*VideoManager) ResolveStream

func (m *VideoManager) ResolveStream(uri string) (*TranscodingRequest, error)

func (*VideoManager) Video

func (m *VideoManager) Video(uri string) (string, error)

Video checks if video exists in the library or waiting in one of the queues. If neither, it adds claim to the pool for later processing.