library

Documentation

Index

• Constants
• Variables
• func GetStreamHasher() hash.Hash
• func PopulateHLSPlaylist(t *testing.T, dstPath, sdHash string)
• func RegisterMetrics()
• func SetLogger(l *zap.SugaredLogger)
• func SpawnLibraryCleaning(lib *Library, storageName string, maxSize uint64) chan struct{}
• func StringToSize(s string) uint64
• func WalkStream(baseURI string, getFn StreamGetter, processFn StreamProcessor) error
• func WithTimestamp(ts time.Time) func(*Manifest)
• func WithVersion(v string) func(*Manifest)
• func WithWorkerName(n string) func(*Manifest)
• type Config
• type Library
  • func New(config Config) *Library
  • func (lib *Library) AddChannel(uri string, priority db.ChannelPriority) (db.Channel, error)
  • func (lib *Library) AddRemoteStream(stream Stream) error
  • func (lib *Library) GetAllChannels() ([]db.Channel, error)
  • func (lib *Library) GetVideo(sdHash string) (db.Video, error)
  • func (lib *Library) GetVideoURL(sdHash string) (string, error)
  • func (lib *Library) Retire(v db.Video) error
  • func (lib *Library) RetireVideos(storageName string, maxSize uint64) (uint64, uint64, error)
  • func (lib *Library) ValidateStreams(storageName string, offset, limit int32, remove bool) ([]string, []string, error)
• type LibraryTestHelper
  • func (h *LibraryTestHelper) SetupLibraryDB() error
  • func (h *LibraryTestHelper) TearDownLibraryDB() error
• type Manifest
• type Storage
• type Stream
  • func GenerateDummyStream() *Stream
  • func InitStream(dir string, remoteStorage string) *Stream
  • func (s *Stream) Checksum() string
  • func (s *Stream) ChecksumValid(checksum string) bool
  • func (s *Stream) GenerateManifest(url, channel, sdHash string, manifestFuncs ...func(*Manifest)) error
  • func (s *Stream) ReadManifest() error
  • func (s *Stream) SDHash() string
  • func (s *Stream) Size() int64
  • func (s *Stream) TID() string
  • func (s *Stream) URL() string
  • func (s *Stream) Walk(walker StreamWalker) error
• type StreamGetter
• type StreamProcessor
• type StreamWalker
• type ValidationResult
  • func ValidateStream(baseURL string, failFast bool, skipSegments bool) (*ValidationResult, error)

Constants

const (
	MasterPlaylistName  = "master.m3u8"
	PlaylistExt         = ".m3u8"
	FragmentExt         = ".ts"
	ManifestName        = ".manifest"
	PlaylistContentType = "application/x-mpegurl"
	FragmentContentType = "video/mp2t"

	SkipChecksum = "SkipChecksumForThisStream"
)

const (
	SchemeRemote = "remote"
)

Variables

var (
	LibraryBytes = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "library_total_bytes",
	})
	LibraryRetiredBytes = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "library_retired_bytes",
	})
)

var ErrStreamNotFound = errors.New("stream not found")

var PopulatedHLSPlaylistFiles = []string{}/* 317 elements not displayed */

var SDHashRe = regexp.MustCompile(`/([A-Za-z0-9]{96})/`)

var SkipSegment = errors.New("skip fragment")

Functions

func GetStreamHasher

func GetStreamHasher() hash.Hash

func PopulateHLSPlaylist

func PopulateHLSPlaylist(t *testing.T, dstPath, sdHash string)

PopulateHLSPlaylist generates a stream of 3131915 bytes in size, segments binary data will all be zeroes.

func RegisterMetrics

func RegisterMetrics()

func SetLogger

func SetLogger(l *zap.SugaredLogger)

func SpawnLibraryCleaning

func SpawnLibraryCleaning(lib *Library, storageName string, maxSize uint64) chan struct{}

func StringToSize

func StringToSize(s string) uint64

func WalkStream

func WalkStream(baseURI string, getFn StreamGetter, processFn StreamProcessor) error

WalkStream parses an HLS playlist, calling `getFn` to load and `processFn` for the master playlist located in `baseURI`, subplaylists and all segments contained within.

func WithTimestamp

func WithTimestamp(ts time.Time) func(*Manifest)

func WithVersion

func WithVersion(v string) func(*Manifest)

func WithWorkerName

func WithWorkerName(n string) func(*Manifest)

Types

type Config

type Config struct {
	Storage Storage
	DB      db.DBTX
	Log     logging.KVLogger
}

type Library

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

func New

func New(config Config) *Library

func (*Library) AddChannel

func (lib *Library) AddChannel(uri string, priority db.ChannelPriority) (db.Channel, error)

func (*Library) AddRemoteStream

func (lib *Library) AddRemoteStream(stream Stream) error

func (*Library) GetAllChannels

func (lib *Library) GetAllChannels() ([]db.Channel, error)

func (*Library) GetVideo

func (lib *Library) GetVideo(sdHash string) (db.Video, error)

func (*Library) GetVideoURL

func (lib *Library) GetVideoURL(sdHash string) (string, error)

func (*Library) Retire

func (lib *Library) Retire(v db.Video) error

func (*Library) RetireVideos

func (lib *Library) RetireVideos(storageName string, maxSize uint64) (uint64, uint64, error)

RetireVideos deletes older videos from S3, keeping total size of remote videos at maxSize.

func (*Library) ValidateStreams

func (lib *Library) ValidateStreams(storageName string, offset, limit int32, remove bool) ([]string, []string, error)

RetireVideos deletes older videos from S3, keeping total size of remote videos at maxSize.

type LibraryTestHelper

type LibraryTestHelper struct {
	DB        *sql.DB
	DBCleanup migrator.TestDBCleanup
}

func (*LibraryTestHelper) SetupLibraryDB

func (h *LibraryTestHelper) SetupLibraryDB() error

func (*LibraryTestHelper) TearDownLibraryDB

func (h *LibraryTestHelper) TearDownLibraryDB() error

type Manifest

type Manifest struct {
	URL        string
	ChannelURL string `yaml:",omitempty" json:"channel_url"`
	SDHash     string

	// Meta attributes
	TranscodedBy string    `yaml:"transcoded_by,omitempty" json:"transcoded_by"`
	TranscodedAt time.Time `yaml:"transcoded_at,omitempty" json:"transcoded_at"`
	Version      string    `yaml:",omitempty"`

	// Auto-filled attributes
	TID      string `yaml:",omitempty"`
	Size     int64  `yaml:",omitempty"`
	Checksum string `yaml:",omitempty"`

	FfmpegArgs string   `yaml:"ffmpeg_args,omitempty"`
	Files      []string `yaml:",omitempty"`
}

type Storage

type Storage interface {
	Name() string
	GetURL(tid string) string
	Put(stream *Stream, _ bool) error
	Delete(streamTID string) error
}

type Stream

type Stream struct {
	LocalPath     string `json:"local_path,omitempty"`
	RemoteStorage string `json:"remote_storage,omitempty"`
	Manifest      *Manifest
}

func GenerateDummyStream

func GenerateDummyStream() *Stream

func InitStream

func InitStream(dir string, remoteStorage string) *Stream

func (*Stream) Checksum

func (s *Stream) Checksum() string

func (*Stream) ChecksumValid

func (s *Stream) ChecksumValid(checksum string) bool

func (*Stream) GenerateManifest

func (s *Stream) GenerateManifest(url, channel, sdHash string, manifestFuncs ...func(*Manifest)) error

GenerateManifest needs to be called for newly initialized (transcoded) streams.

func (*Stream) ReadManifest

func (s *Stream) ReadManifest() error

func (*Stream) SDHash

func (s *Stream) SDHash() string

func (*Stream) Size

func (s *Stream) Size() int64

func (*Stream) TID

func (s *Stream) TID() string

func (*Stream) URL

func (s *Stream) URL() string

func (*Stream) Walk

func (s *Stream) Walk(walker StreamWalker) error

type StreamGetter

type StreamGetter func(path ...string) (io.ReadCloser, error)

type StreamProcessor

type StreamProcessor func(fgName string, r io.ReadCloser) error

type StreamWalker

type StreamWalker func(fi fs.FileInfo, fullPath, name string) error

type ValidationResult

type ValidationResult struct {
	URL              string
	Present, Missing []string
}

func ValidateStream

func ValidateStream(baseURL string, failFast bool, skipSegments bool) (*ValidationResult, error)