utils

Documentation

Index

• Constants
• Variables
• func EnableLockTracker()
• func HashedID(id string) string
• func LocalNodeID() (string, error)
• func NewGuid(prefix string) string
• func NumMutexes() int
• func RandomSecret() string
• func ScanTrackedLocks(threshold time.Duration) bool
• func ScanTrackedLocksI(threshold time.Duration, n int) bool
• type CPUStats
  • func NewCPUStats(updateCallback func(idle float64)) (*CPUStats, error)
  • func (c *CPUStats) GetCPUIdle() float64
  • func (c *CPUStats) NumCPU() int
  • func (c *CPUStats) Stop()
• type MessageBus
  • func NewRedisMessageBus(rc redis.UniversalClient) MessageBus
• type Mutex
  • func (m *Mutex) Lock()
• type PubSub
• type RWMutex
  • func (m *RWMutex) Lock()
  • func (m *RWMutex) RLock()
• type RedisMessageBus
  • func (r *RedisMessageBus) Lock(ctx context.Context, key string, expiration time.Duration) (bool, error)
  • func (r *RedisMessageBus) Publish(ctx context.Context, channel string, msg proto.Message) error
  • func (r *RedisMessageBus) Subscribe(ctx context.Context, channel string) (PubSub, error)
  • func (r *RedisMessageBus) SubscribeQueue(ctx context.Context, channel string) (PubSub, error)
• type RedisPubSub
  • func (r *RedisPubSub) Channel() <-chan interface{}
  • func (r *RedisPubSub) Close() error
  • func (r *RedisPubSub) Payload(msg interface{}) []byte

Constants

const (
	RoomPrefix        = "RM_"
	NodePrefix        = "ND_"
	ParticipantPrefix = "PA_"
	TrackPrefix       = "TR_"
	APIKeyPrefix      = "API"
	EgressPrefix      = "EG_"
	IngressPrefix     = "IN_"
	RPCPrefix         = "RPC_"
)

Variables

var (
	PromMessageBusCounter = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "livekit",
			Subsystem: "messagebus",
			Name:      "messages",
		},
		[]string{"type", "status"},
	)
)

Functions

func EnableLockTracker

func EnableLockTracker()

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

func HashedID

func HashedID(id string) string

Creates a hashed ID from a unique string

func LocalNodeID

func LocalNodeID() (string, error)

func NewGuid

func NewGuid(prefix string) string

func NumMutexes

func NumMutexes() int

func RandomSecret

func RandomSecret() string

func ScanTrackedLocks

func ScanTrackedLocks(threshold time.Duration) bool

ScanTrackedLocks check all lock trackers

func ScanTrackedLocksI

func ScanTrackedLocksI(threshold time.Duration, n int) bool

ScanTrackedLocksI check lock trackers incrementally n at a time

Types

type CPUStats

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

func NewCPUStats

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

func (*CPUStats) GetCPUIdle

func (c *CPUStats) GetCPUIdle() float64

func (*CPUStats) NumCPU

func (c *CPUStats) NumCPU() int

func (*CPUStats) Stop

func (c *CPUStats) Stop()

type MessageBus

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

func NewRedisMessageBus(rc redis.UniversalClient) MessageBus

type Mutex

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

func (*Mutex) Lock

func (m *Mutex) Lock()

type PubSub

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

type RWMutex

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

func (*RWMutex) Lock

func (m *RWMutex) Lock()

func (*RWMutex) RLock

func (m *RWMutex) RLock()

type RedisMessageBus

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

func (*RedisMessageBus) Lock

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

func (*RedisMessageBus) Publish

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

func (*RedisMessageBus) Subscribe

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

func (*RedisMessageBus) SubscribeQueue

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

type RedisPubSub

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

func (*RedisPubSub) Channel

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

func (*RedisPubSub) Close

func (r *RedisPubSub) Close() error

func (*RedisPubSub) Payload

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