utils

package
v1.6.1 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2023 License: Apache-2.0 Imports: 38 Imported by: 48

Documentation

Index

Constants

View Source 
const (
	RoomPrefix         = "RM_"
	NodePrefix         = "ND_"
	ParticipantPrefix  = "PA_"
	TrackPrefix        = "TR_"
	APIKeyPrefix       = "API"
	EgressPrefix       = "EG_"
	IngressPrefix      = "IN_"
	RPCPrefix          = "RPC_"
	WHIPResourcePrefix = "WH_"
	RTMPResourcePrefix = "RT_"
	URLResourcePrefix  = "UR_"
)

Variables

View Source 
var (
	ErrAnachronousSample = errors.New("anachronous sample")
)
View Source 
var (
	PromMessageBusCounter = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "livekit",
			Subsystem: "messagebus",
			Name:      "messages",
		},
		[]string{"type", "status"},
	)
)

Functions

func EnableLockTracker added in v1.4.0 

func EnableLockTracker()

EnableLockTracker enable lock tracking background worker. This should be called during init

func GetMimeTypeForAudioCodec added in v1.5.1 

func GetMimeTypeForAudioCodec(codec livekit.AudioCodec) string

func GetMimeTypeForVideoCodec added in v1.5.1 

func GetMimeTypeForVideoCodec(codec livekit.VideoCodec) string

func HashedID 

func HashedID(id string) string

Creates a hashed ID from a unique string

func Least added in v1.5.9 

func Least[T Numeric](less func(a, b T) bool, vs ...T) T

func LocalNodeID added in v0.11.1 

func LocalNodeID() (string, error)

func MarshalGuid added in v1.5.7 

func MarshalGuid[T livekit.Guid](id T) livekit.GuidBlock

func Max added in v1.5.9 

func Max[T Numeric](vs ...T) T

func Min added in v1.5.9 

func Min[T Numeric](vs ...T) T

func Most added in v1.5.9 

func Most[T Numeric](less func(a, b T) bool, vs ...T) T

func NewGuid 

func NewGuid(prefix string) string

func NumMutexes added in v1.4.0 

func NumMutexes() int

func ParallelExec added in v1.4.3 

func ParallelExec[T any](vals []T, parallelThreshold, step uint64, fn func(T))

ParallelExec will executes the given function with each element of vals, if len(vals) >= parallelThreshold, will execute them in parallel, with the given step size. So fn must be thread-safe.

func RandomSecret 

func RandomSecret() string

func Redact added in v1.5.10 

func Redact(s, name string) string

func RedactIdentifier added in v1.5.10 

func RedactIdentifier(identifier string) string

func RedactStreamKey added in v1.5.10 

func RedactStreamKey(url string) (string, bool)

func UnmarshalGuid added in v1.5.7 

func UnmarshalGuid[T livekit.Guid](b livekit.GuidBlock) T

Types

type CPUStats added in v1.2.0 

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

func NewCPUStats added in v1.2.0 

func NewCPUStats(updateCallback func(idle float64)) (*CPUStats, error)

func (*CPUStats) GetCPUIdle added in v1.2.0 

func (c *CPUStats) GetCPUIdle() float64

func (*CPUStats) NumCPU added in v1.2.3 

func (c *CPUStats) NumCPU() float64

func (*CPUStats) Stop added in v1.2.0 

func (c *CPUStats) Stop()

type DedupedSlice added in v1.6.0 

type DedupedSlice[T comparable] struct {
	// contains filtered or unexported fields
}

func NewDedupedSlice added in v1.6.0 

func NewDedupedSlice[T comparable](maxLen int) *DedupedSlice[T]

func (*DedupedSlice[T]) Add added in v1.6.0 

func (d *DedupedSlice[T]) Add(val T) bool

func (*DedupedSlice[T]) Clear added in v1.6.0 

func (d *DedupedSlice[T]) Clear()

func (*DedupedSlice[T]) Get added in v1.6.0 

func (d *DedupedSlice[T]) Get() []T

func (*DedupedSlice[T]) Has added in v1.6.0 

func (d *DedupedSlice[T]) Has(val T) bool

func (*DedupedSlice[T]) Len added in v1.6.0 

func (d *DedupedSlice[T]) Len() int

type ErrArray added in v1.5.6 

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

func (*ErrArray) AppendErr added in v1.5.6 

func (e *ErrArray) AppendErr(err error)

func (*ErrArray) ToError added in v1.5.6 

func (e *ErrArray) ToError() psrpc.Error

type EventEmitter added in v1.5.7 

type EventEmitter[K comparable, V any] struct {
	// contains filtered or unexported fields
}

func NewDefaultEventEmitter added in v1.5.7 

func NewDefaultEventEmitter[K comparable, V any]() *EventEmitter[K, V]

func NewEventEmitter added in v1.5.7 

func NewEventEmitter[K comparable, V any](params EventEmitterParams) *EventEmitter[K, V]

func (*EventEmitter[K, V]) Emit added in v1.5.7 

func (e *EventEmitter[K, V]) Emit(k K, v V)

func (*EventEmitter[K, V]) Observe added in v1.5.7 

func (e *EventEmitter[K, V]) Observe(k K) *EventObserver[V]

func (*EventEmitter[K, V]) ObservedKeys added in v1.5.7 

func (e *EventEmitter[K, V]) ObservedKeys() []K

type EventEmitterParams added in v1.5.7 

type EventEmitterParams struct {
	QueueSize int
	Logger    logger.Logger
}

func DefaultEventEmitterParams added in v1.5.8 

func DefaultEventEmitterParams() EventEmitterParams

type EventObserver added in v1.5.7 

type EventObserver[V any] struct {
	// contains filtered or unexported fields
}

func NewEventObserver added in v1.5.7 

func NewEventObserver[V any](stopFunc func()) (*EventObserver[V], func(v V))

func (*EventObserver[V]) Events added in v1.5.7 

func (o *EventObserver[V]) Events() <-chan V

func (*EventObserver[V]) Stop added in v1.5.7 

func (o *EventObserver[V]) Stop()

type EventObserverList added in v1.5.8 

type EventObserverList[V any] struct {
	// contains filtered or unexported fields
}

func NewDefaultEventObserverList added in v1.5.8 

func NewDefaultEventObserverList[V any]() *EventObserverList[V]

func NewEventObserverList added in v1.5.8 

func NewEventObserverList[V any](params EventEmitterParams) *EventObserverList[V]

func (*EventObserverList[V]) Emit added in v1.5.8 

func (l *EventObserverList[V]) Emit(v V)

func (*EventObserverList[V]) Len added in v1.5.8 

func (l *EventObserverList[V]) Len() int

func (*EventObserverList[V]) Observe added in v1.5.8 

func (l *EventObserverList[V]) Observe() *EventObserver[V]

type FlowGraph added in v1.5.7 

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

func NewFlowGraph added in v1.5.7 

func NewFlowGraph(n int64) FlowGraph

func (*FlowGraph) AddEdge added in v1.5.7 

func (g *FlowGraph) AddEdge(s, t, cap, cost int64)

type Graph added in v1.5.7 

type Graph[K comparable, N GraphNodeProps[K], E GraphEdgeProps] struct {
	// contains filtered or unexported fields
}

func NewGraph added in v1.5.7 

func NewGraph[K comparable, N GraphNodeProps[K], E GraphEdgeProps]() *Graph[K, N, E]

func (*Graph[K, N, E]) DeleteEdge added in v1.5.7 

func (g *Graph[K, N, E]) DeleteEdge(src, dst K)

func (*Graph[K, N, E]) DeleteNode added in v1.5.8 

func (g *Graph[K, N, E]) DeleteNode(id K)

func (*Graph[K, N, E]) Edge added in v1.5.7 

func (g *Graph[K, N, E]) Edge(src, dst K) (p E)

func (*Graph[K, N, E]) HasEdge added in v1.5.8 

func (g *Graph[K, N, E]) HasEdge(src, dst K) bool

func (*Graph[K, N, E]) HasNode added in v1.5.8 

func (g *Graph[K, N, E]) HasNode(id K) bool

func (*Graph[K, N, E]) InEdges added in v1.5.7 

func (g *Graph[K, N, E]) InEdges(dst K) map[K]E

func (*Graph[K, N, E]) InsertEdge added in v1.5.7 

func (g *Graph[K, N, E]) InsertEdge(src, dst K, props E)

func (*Graph[K, N, E]) InsertNode added in v1.5.7 

func (g *Graph[K, N, E]) InsertNode(props N)

func (*Graph[K, N, E]) Node added in v1.5.7 

func (g *Graph[K, N, E]) Node(id K) (props N)

func (*Graph[K, N, E]) OutEdges added in v1.5.7 

func (g *Graph[K, N, E]) OutEdges(src K) map[K]E

func (*Graph[K, N, E]) ShortestPath added in v1.5.7 

func (g *Graph[K, N, E]) ShortestPath(src, dst K) ([]N, int64)

func (*Graph[K, N, E]) Size added in v1.5.7 

func (g *Graph[K, N, E]) Size() int

func (*Graph[K, N, E]) TopologicalSort added in v1.5.7 

func (g *Graph[K, N, E]) TopologicalSort() []N

type GraphEdge added in v1.5.7 

type GraphEdge[N, E any] struct {
	// contains filtered or unexported fields
}

type GraphEdgeProps added in v1.5.7 

type GraphEdgeProps interface {
	Length() int64
}

type GraphNode added in v1.5.7 

type GraphNode[T any] struct {
	// contains filtered or unexported fields
}

type GraphNodeProps added in v1.5.7 

type GraphNodeProps[K comparable] interface {
	ID() K
}

type MessageBus added in v0.8.4 

type MessageBus interface {
	Subscribe(ctx context.Context, channel string) (PubSub, error)
	// SubscribeQueue is like subscribe, but ensuring only a single instance gets to process the message
	SubscribeQueue(ctx context.Context, channel string) (PubSub, error)
	Publish(ctx context.Context, channel string, msg proto.Message) error
}

func NewRedisMessageBus added in v0.8.4 

func NewRedisMessageBus(rc redis.UniversalClient) MessageBus

type MinCostMaxFlow added in v1.5.7 

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

func (*MinCostMaxFlow) ComputeMaxFlow added in v1.5.7 

func (f *MinCostMaxFlow) ComputeMaxFlow(g FlowGraph, s, t int64) (flow, cost int64)

func (*MinCostMaxFlow) Flow added in v1.5.7 

func (f *MinCostMaxFlow) Flow(s, t int64) int64

type Mutex added in v1.4.0 

type Mutex struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func (*Mutex) Lock added in v1.4.0 

func (m *Mutex) Lock()

func (*Mutex) Unlock added in v1.5.7 

func (m *Mutex) Unlock()

type Numeric added in v1.5.9 

type Numeric interface {
	constraints.Signed | constraints.Unsigned | time.Duration
}

type ProtoProxy added in v1.5.5 

type ProtoProxy[T proto.Message] struct {
	// contains filtered or unexported fields
}

ProtoProxy is a caching proxy for protobuf messages that may be expensive to compute. It is used to avoid unnecessary re-generation of Protobufs

func NewProtoProxy added in v1.5.5 

func NewProtoProxy[T proto.Message](refreshInterval time.Duration, updateFn func() T) *ProtoProxy[T]

NewProtoProxy creates a new ProtoProxy that regenerates underlying values at a cadence of refreshInterval this should be used for updates that should be sent periodically, but does not have the urgency of immediate delivery updateFn should provide computations required to generate the protobuf if refreshInterval is 0, then proxy will only update on MarkDirty(true)

func (*ProtoProxy[T]) Get added in v1.5.5 

func (p *ProtoProxy[T]) Get() T

func (*ProtoProxy[T]) MarkDirty added in v1.5.5 

func (p *ProtoProxy[T]) MarkDirty(immediate bool)

func (*ProtoProxy[T]) Stop added in v1.5.5 

func (p *ProtoProxy[T]) Stop()

func (*ProtoProxy[T]) Updated added in v1.5.5 

func (p *ProtoProxy[T]) Updated() <-chan struct{}

type PubSub added in v0.8.4 

type PubSub interface {
	Channel() <-chan interface{}
	Payload(msg interface{}) []byte
	Close() error
}

type RWMutex added in v1.4.0 

type RWMutex struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*RWMutex) Lock added in v1.4.0 

func (m *RWMutex) Lock()

func (*RWMutex) RLock added in v1.4.0 

func (m *RWMutex) RLock()

func (*RWMutex) RUnlock added in v1.5.7 

func (m *RWMutex) RUnlock()

func (*RWMutex) Unlock added in v1.5.7 

func (m *RWMutex) Unlock()

type RedisMessageBus added in v0.8.4 

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

func (*RedisMessageBus) Lock added in v0.8.4 

func (r *RedisMessageBus) Lock(ctx context.Context, key string, expiration time.Duration) (bool, error)

func (*RedisMessageBus) Publish added in v0.8.4 

func (r *RedisMessageBus) Publish(ctx context.Context, channel string, msg proto.Message) error

func (*RedisMessageBus) Subscribe added in v0.8.4 

func (r *RedisMessageBus) Subscribe(ctx context.Context, channel string) (PubSub, error)

func (*RedisMessageBus) SubscribeQueue added in v0.9.3 

func (r *RedisMessageBus) SubscribeQueue(ctx context.Context, channel string) (PubSub, error)

type RedisPubSub added in v0.8.4 

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

func (*RedisPubSub) Channel added in v0.8.4 

func (r *RedisPubSub) Channel() <-chan interface{}

func (*RedisPubSub) Close added in v0.8.4 

func (r *RedisPubSub) Close() error

func (*RedisPubSub) Payload added in v0.8.4 

func (r *RedisPubSub) Payload(msg interface{}) []byte

type StuckLock added in v1.5.7 

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

func ScanTrackedLocks added in v1.4.0 

func ScanTrackedLocks(threshold time.Duration) []*StuckLock

ScanTrackedLocks check all lock trackers

func ScanTrackedLocksI added in v1.4.0 

func ScanTrackedLocksI(threshold time.Duration, n int) []*StuckLock

ScanTrackedLocksI check lock trackers incrementally n at a time

func (*StuckLock) FirstLockedAtStack added in v1.5.7 

func (d *StuckLock) FirstLockedAtStack() string

func (*StuckLock) HeldSince added in v1.5.7 

func (d *StuckLock) HeldSince() time.Time

func (*StuckLock) NumGoroutineHeld added in v1.5.7 

func (d *StuckLock) NumGoroutineHeld() int

func (*StuckLock) NumGoroutineWaiting added in v1.5.7 

func (d *StuckLock) NumGoroutineWaiting() int

type TimedAggregator added in v1.5.9 

type TimedAggregator[T number] struct {
	// contains filtered or unexported fields
}

func NewTimedAggregator added in v1.5.9 

func NewTimedAggregator[T number](params TimedAggregatorParams) *TimedAggregator[T]

func (*TimedAggregator[T]) AddSample added in v1.5.9 

func (t *TimedAggregator[T]) AddSample(val T) error

func (*TimedAggregator[T]) AddSampleAt added in v1.5.9 

func (t *TimedAggregator[T]) AddSampleAt(val T, at time.Time) error

func (*TimedAggregator[T]) GetAggregate added in v1.5.9 

func (t *TimedAggregator[T]) GetAggregate() (T, time.Duration)

func (*TimedAggregator[T]) GetAggregateAndRestartAt added in v1.5.9 

func (t *TimedAggregator[T]) GetAggregateAndRestartAt(at time.Time) (T, time.Duration, error)

func (*TimedAggregator[T]) GetAggregateAt added in v1.5.9 

func (t *TimedAggregator[T]) GetAggregateAt(at time.Time) (T, time.Duration, error)

func (*TimedAggregator[T]) GetAverage added in v1.5.9 

func (t *TimedAggregator[T]) GetAverage() float64

func (*TimedAggregator[T]) GetAverageAndRestartAt added in v1.5.9 

func (t *TimedAggregator[T]) GetAverageAndRestartAt(at time.Time) (float64, error)

func (*TimedAggregator[T]) GetAverageAt added in v1.5.9 

func (t *TimedAggregator[T]) GetAverageAt(at time.Time) (float64, error)

func (*TimedAggregator[T]) Reset added in v1.5.9 

func (t *TimedAggregator[T]) Reset()

func (*TimedAggregator[T]) Restart added in v1.5.9 

func (t *TimedAggregator[T]) Restart()

func (*TimedAggregator[T]) RestartAt added in v1.5.9 

func (t *TimedAggregator[T]) RestartAt(at time.Time)

type TimedAggregatorParams added in v1.5.9 

type TimedAggregatorParams struct {
	CapNegativeValues bool
}

type TimedVersion added in v1.4.3 

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

func NewTimedVersionFromProto added in v1.4.3 

func NewTimedVersionFromProto(proto *livekit.TimedVersion) *TimedVersion

func NewTimedVersionFromTime added in v1.4.3 

func NewTimedVersionFromTime(t time.Time) *TimedVersion

func TimedVersionFromProto added in v1.4.3 

func TimedVersionFromProto(proto *livekit.TimedVersion) TimedVersion

func TimedVersionFromTime added in v1.4.3 

func TimedVersionFromTime(t time.Time) TimedVersion

func (*TimedVersion) After added in v1.4.3 

func (t *TimedVersion) After(other *TimedVersion) bool

func (*TimedVersion) Compare added in v1.4.3 

func (t *TimedVersion) Compare(other *TimedVersion) int

func (*TimedVersion) IsZero added in v1.5.5 

func (t *TimedVersion) IsZero() bool

func (*TimedVersion) Load added in v1.4.3 

func (t *TimedVersion) Load() TimedVersion

func (*TimedVersion) Store added in v1.4.3 

func (t *TimedVersion) Store(other *TimedVersion)

func (*TimedVersion) String added in v1.4.3 

func (t *TimedVersion) String() string

func (*TimedVersion) Time added in v1.4.3 

func (t *TimedVersion) Time() time.Time

func (*TimedVersion) ToProto added in v1.4.3 

func (t *TimedVersion) ToProto() *livekit.TimedVersion

func (*TimedVersion) Update added in v1.4.3 

func (t *TimedVersion) Update(other *TimedVersion) bool

type TimedVersionGenerator added in v1.4.3 

type TimedVersionGenerator interface {
	New() *TimedVersion
	Next() TimedVersion
}

func NewDefaultTimedVersionGenerator added in v1.4.3 

func NewDefaultTimedVersionGenerator() TimedVersionGenerator

type TimeoutQueue added in v1.6.0 

type TimeoutQueue[T any] struct {
	// contains filtered or unexported fields
}

func (*TimeoutQueue[T]) IterateAfter added in v1.6.0 

func (q *TimeoutQueue[T]) IterateAfter(timeout time.Duration) *timeoutQueueIterator[T]

func (*TimeoutQueue[T]) IterateRemoveAfter added in v1.6.0 

func (q *TimeoutQueue[T]) IterateRemoveAfter(timeout time.Duration) *timeoutQueueIterator[T]

func (*TimeoutQueue[T]) Remove added in v1.6.0 

func (q *TimeoutQueue[T]) Remove(i *TimeoutQueueItem[T])

func (*TimeoutQueue[T]) Reset added in v1.6.0 

func (q *TimeoutQueue[T]) Reset(i *TimeoutQueueItem[T]) bool

type TimeoutQueueItem added in v1.6.0 

type TimeoutQueueItem[T any] struct {
	Value T
	// contains filtered or unexported fields
}

type Welford added in v1.5.9 

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

Welford implements Welford's online algorithm for variance SEE: https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance#Welford's_online_algorithm

func WelfordMerge added in v1.5.9 

func WelfordMerge(ws ...Welford) Welford

func (Welford) Count added in v1.5.9 

func (w Welford) Count() float64

func (Welford) Mean added in v1.5.9 

func (w Welford) Mean() float64

func (*Welford) Reset added in v1.5.9 

func (w *Welford) Reset()

func (Welford) StdDev added in v1.5.9 

func (w Welford) StdDev() float64

func (*Welford) Update added in v1.5.9 

func (w *Welford) Update(v float64)

func (Welford) Value added in v1.5.9 

func (w Welford) Value() (mean, variance, sampleVariance float64)

func (Welford) Variance added in v1.5.9 

func (w Welford) Variance() float64

type WorkerGroup added in v1.5.2 

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

func (*WorkerGroup) Go added in v1.5.2 

func (w *WorkerGroup) Go(fn func())

func (*WorkerGroup) Wait added in v1.5.2 

func (w *WorkerGroup) Wait()

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.