worker

package
v0.0.118 Latest Latest
Warning

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

 Go to latest Published: Jun 14, 2026 License: MIT Imports: 47 Imported by: 0

Documentation

Index

Constants

View Source 
const PreConnectAudioBufferStream = "lk.agent.pre-connect-audio-buffer"
View Source 
const RoomIOAgentStateAttribute = "lk.agent.state"
View Source 
const RoomIOChatTopic = "lk.chat"
View Source 
const RoomIOPublishOnBehalfAttribute = "lk.publish_on_behalf"
View Source 
const RoomIOSimulatorAttribute = "lk.simulator"
View Source 
const RoomIOTranscriptionFinalAttribute = "lk.transcription_final"
View Source 
const RoomIOTranscriptionSegmentIDAttribute = "lk.segment_id"
View Source 
const RoomIOTranscriptionTopic = "lk.transcription"
View Source 
const RoomIOTranscriptionTrackIDAttribute = "lk.transcribed_track_id"
View Source 
const WorkerProtocolVersion = 1

Variables

This section is empty.

Functions

This section is empty.

Types

type AgentServer 

type AgentServer struct {
	Options WorkerOptions
	// contains filtered or unexported fields
}

func NewAgentServer 

func NewAgentServer(opts WorkerOptions) *AgentServer

func (*AgentServer) ActiveJobs added in v0.0.15 

func (s *AgentServer) ActiveJobs() []*JobContext

func (*AgentServer) ActiveRunningJobs added in v0.0.15 

func (s *AgentServer) ActiveRunningJobs() []workeripc.RunningJobInfo

func (*AgentServer) Drain added in v0.0.5 

func (s *AgentServer) Drain(ctx context.Context) error

func (*AgentServer) DrainWithTimeout added in v0.0.15 

func (s *AgentServer) DrainWithTimeout(ctx context.Context, timeout time.Duration) error

func (*AgentServer) Draining added in v0.0.15 

func (s *AgentServer) Draining() bool

func (*AgentServer) ExecuteLocalJob 

func (s *AgentServer) ExecuteLocalJob(ctx context.Context, roomName string, participantIdentity string) error

ExecuteLocalJob runs a job locally without connecting to the worker service, useful for the CLI console

func (*AgentServer) ExecuteLocalJobWithOptions added in v0.0.15 

func (s *AgentServer) ExecuteLocalJobWithOptions(ctx context.Context, roomName string, participantIdentity string, options LocalJobOptions) error

func (*AgentServer) ExecuteRunningJob added in v0.0.15 

func (s *AgentServer) ExecuteRunningJob(ctx context.Context, info workeripc.RunningJobInfo) error

func (*AgentServer) GetConsoleSession 

func (s *AgentServer) GetConsoleSession() any

GetConsoleSession retrieves the active local console session

func (*AgentServer) ID added in v0.0.15 

func (s *AgentServer) ID() string

func (*AgentServer) OnWorkerRegistered added in v0.0.15 

func (s *AgentServer) OnWorkerRegistered(handler WorkerRegisteredHandler)

func (*AgentServer) OnWorkerStarted added in v0.0.15 

func (s *AgentServer) OnWorkerStarted(handler WorkerStartedHandler)

func (*AgentServer) RTCSession 

func (s *AgentServer) RTCSession(
	entrypoint func(*JobContext) error,
	request func(*JobRequest) error,
	sessionEnd func(*JobContext) error,
) error

func (*AgentServer) ReloadRunningJobs added in v0.0.15 

func (s *AgentServer) ReloadRunningJobs(ctx context.Context, jobs []workeripc.RunningJobInfo, now time.Time) error

func (*AgentServer) Run 

func (s *AgentServer) Run(ctx context.Context) error

func (*AgentServer) RunUnregistered added in v0.0.15 

func (s *AgentServer) RunUnregistered(ctx context.Context) error

func (*AgentServer) SetConsoleSession 

func (s *AgentServer) SetConsoleSession(session any)

SetConsoleSession allows entrypoints to register their session for console interaction

func (*AgentServer) UpdateOptions added in v0.0.15 

func (s *AgentServer) UpdateOptions(opts WorkerOptions) error

func (*AgentServer) WorkerInfo added in v0.0.15 

func (s *AgentServer) WorkerInfo() WorkerInfo

type AudioDecoder 

type AudioDecoder interface {
	Decode(data []byte) ([]byte, error)
	Close() error
}

type AudioEncoder 

type AudioEncoder interface {
	Encode(pcm []byte) ([]byte, error)
	Close() error
}

type AutoSubscribe added in v0.0.15 

type AutoSubscribe string
const (
	AutoSubscribeSubscribeAll  AutoSubscribe = "subscribe_all"
	AutoSubscribeSubscribeNone AutoSubscribe = "subscribe_none"
	AutoSubscribeAudioOnly     AutoSubscribe = "audio_only"
	AutoSubscribeVideoOnly     AutoSubscribe = "video_only"
)

type ConnectOptions added in v0.0.15 

type ConnectOptions struct {
	AutoSubscribe AutoSubscribe
}

type JobAPI added in v0.0.15 

type JobAPI struct {
	RoomService JobRoomServiceAPI
	SIP         JobSIPAPI
}

func NewJobAPI added in v0.0.15 

func NewJobAPI(url string, apiKey string, apiSecret string) *JobAPI

type JobAcceptArguments 

type JobAcceptArguments struct {
	Name       string
	Identity   string
	Metadata   string
	Attributes map[string]string
}

type JobContext 

type JobContext struct {
	Job             *livekit.Job
	Room            *lksdk.Room
	Report          *agent.SessionReport
	AcceptArguments JobAcceptArguments
	// contains filtered or unexported fields
}

func GetCurrentJobContext added in v0.0.15 

func GetCurrentJobContext() (*JobContext, bool)

GetCurrentJobContext is an alias for GetJobContext kept for reference parity with LiveKit Agents' get_current_job_context name.

func GetJobContext added in v0.0.15 

func GetJobContext() (*JobContext, bool)

GetJobContext returns the JobContext currently executing on this goroutine.

This mirrors LiveKit Agents' get_job_context helper for code that runs inside a worker job entrypoint. Go does not have Python contextvars, so newly spawned goroutines should receive the JobContext explicitly instead of relying on this helper.

func NewJobContext 

func NewJobContext(job *livekit.Job, url string, apiKey string, apiSecret string) *JobContext

func RequireCurrentJobContext added in v0.0.15 

func RequireCurrentJobContext() (*JobContext, error)

func RequireJobContext added in v0.0.15 

func RequireJobContext() (*JobContext, error)

func (*JobContext) API added in v0.0.15 

func (c *JobContext) API() *JobAPI

func (*JobContext) AddParticipantEntrypoint added in v0.0.15 

func (c *JobContext) AddParticipantEntrypoint(entrypoint ParticipantEntrypoint, kinds ...livekit.ParticipantInfo_Kind) error

func (*JobContext) AddSIPParticipant 

func (c *JobContext) AddSIPParticipant(ctx context.Context, callTo string, trunkID string, identity string, names ...string) (*livekit.SIPParticipantInfo, error)

AddSIPParticipant adds a SIP participant to the room.

func (*JobContext) AddShutdownCallback added in v0.0.15 

func (c *JobContext) AddShutdownCallback(callback any) error

func (*JobContext) Agent added in v0.0.15 

func (c *JobContext) Agent() *lksdk.LocalParticipant

func (*JobContext) AvatarStartInfo added in v0.0.15 

func (c *JobContext) AvatarStartInfo() agent.AvatarStartInfo

func (*JobContext) Connect 

func (c *JobContext) Connect(ctx context.Context, cb *lksdk.RoomCallback, options ...ConnectOptions) error

func (*JobContext) CreateSIPParticipant added in v0.0.15 

func (c *JobContext) CreateSIPParticipant(ctx context.Context, req *livekit.CreateSIPParticipantRequest) (*livekit.SIPParticipantInfo, error)

func (*JobContext) DeleteRoom 

func (c *JobContext) DeleteRoom(ctx context.Context, roomName string) (*livekit.DeleteRoomResponse, error)

DeleteRoom deletes the room and disconnects all participants.

func (*JobContext) InitRecording added in v0.0.15 

func (c *JobContext) InitRecording(options agent.RecordingOptions)

func (*JobContext) IsFakeJob added in v0.0.15 

func (c *JobContext) IsFakeJob() bool

func (*JobContext) JobID added in v0.0.15 

func (c *JobContext) JobID() string

func (*JobContext) LocalParticipantIdentity added in v0.0.15 

func (c *JobContext) LocalParticipantIdentity() string

func (*JobContext) LogContextFields added in v0.0.15 

func (c *JobContext) LogContextFields() map[string]any

func (*JobContext) MakeSessionReport added in v0.0.15 

func (c *JobContext) MakeSessionReport(sessions ...*agent.AgentSession) (*agent.SessionReport, error)

func (*JobContext) MoveParticipant added in v0.0.15 

func (c *JobContext) MoveParticipant(ctx context.Context, room string, identity string, destinationRoom string) error

func (*JobContext) ParticipantIdentity added in v0.0.15 

func (c *JobContext) ParticipantIdentity() string

func (*JobContext) PrimarySession added in v0.0.15 

func (c *JobContext) PrimarySession() (*agent.AgentSession, error)

func (*JobContext) Proc added in v0.0.15 

func (c *JobContext) Proc() *JobProcess

func (*JobContext) PublisherInfo added in v0.0.15 

func (c *JobContext) PublisherInfo() *livekit.ParticipantInfo

func (*JobContext) RoomInfo added in v0.0.15 

func (c *JobContext) RoomInfo() *livekit.Room

func (*JobContext) SessionDirectory added in v0.0.11 

func (c *JobContext) SessionDirectory() string

func (*JobContext) SetLogContextFields added in v0.0.15 

func (c *JobContext) SetLogContextFields(fields map[string]any)

func (*JobContext) SetPrimarySession added in v0.0.15 

func (c *JobContext) SetPrimarySession(session *agent.AgentSession)

func (*JobContext) SetSessionDirectory added in v0.0.15 

func (c *JobContext) SetSessionDirectory(path string)

func (*JobContext) Shutdown 

func (c *JobContext) Shutdown(reasons ...string)

func (*JobContext) ShutdownDone added in v0.0.18 

func (c *JobContext) ShutdownDone() <-chan struct{}

func (*JobContext) Tagger added in v0.0.15 

func (c *JobContext) Tagger() *agent.Tagger

func (*JobContext) Terminated added in v0.0.18 

func (c *JobContext) Terminated() bool

func (*JobContext) TokenClaims added in v0.0.15 

func (c *JobContext) TokenClaims() (*auth.ClaimGrants, error)

func (*JobContext) TransferSIPParticipant 

func (c *JobContext) TransferSIPParticipant(ctx context.Context, identity string, transferTo string, playDialtones ...bool) error

TransferSIPParticipant transfers a SIP participant to another number.

func (*JobContext) TransferSIPParticipantByParticipant added in v0.0.15 

func (c *JobContext) TransferSIPParticipantByParticipant(ctx context.Context, participant any, transferTo string, playDialtones ...bool) error

func (*JobContext) WaitForAgent added in v0.0.15 

func (c *JobContext) WaitForAgent(
	ctx context.Context,
	agentName ...string,
) (*lksdk.RemoteParticipant, error)

func (*JobContext) WaitForParticipant added in v0.0.15 

func (c *JobContext) WaitForParticipant(
	ctx context.Context,
	identity string,
	kinds ...livekit.ParticipantInfo_Kind,
) (*lksdk.RemoteParticipant, error)

func (*JobContext) WaitForParticipantAttribute added in v0.0.15 

func (c *JobContext) WaitForParticipantAttribute(
	ctx context.Context,
	identity string,
	attribute string,
	value string,
) error

func (*JobContext) WaitForTrackPublication added in v0.0.15 

func (c *JobContext) WaitForTrackPublication(
	ctx context.Context,
	identity string,
	kinds ...livekit.TrackType,
) (*lksdk.RemoteTrackPublication, error)

func (*JobContext) WaitForTrackPublicationWithOptions added in v0.0.15 

func (c *JobContext) WaitForTrackPublicationWithOptions(
	ctx context.Context,
	options utils.TrackPublicationWaitOptions,
) (*lksdk.RemoteTrackPublication, error)

func (*JobContext) WorkerID added in v0.0.15 

func (c *JobContext) WorkerID() string

type JobExecutorType added in v0.0.15 

type JobExecutorType = workeripc.ExecutorType
const (
	JobExecutorTypeThread  JobExecutorType = workeripc.ExecutorTypeThread
	JobExecutorTypeProcess JobExecutorType = workeripc.ExecutorTypeProcess
)

type JobProcess added in v0.0.15 

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

func NewJobProcess added in v0.0.15 

func NewJobProcess(executorType JobExecutorType, userArguments any, httpProxy string) *JobProcess

func (*JobProcess) ExecutorType added in v0.0.15 

func (p *JobProcess) ExecutorType() JobExecutorType

func (*JobProcess) HTTPProxy added in v0.0.15 

func (p *JobProcess) HTTPProxy() string

func (*JobProcess) PID added in v0.0.15 

func (p *JobProcess) PID() int

func (*JobProcess) UserArguments added in v0.0.15 

func (p *JobProcess) UserArguments() any

func (*JobProcess) Userdata added in v0.0.15 

func (p *JobProcess) Userdata() map[any]any

type JobRejectArguments added in v0.0.15 

type JobRejectArguments struct {
	Terminate bool
}

type JobRequest 

type JobRequest struct {
	Job *livekit.Job
	// contains filtered or unexported fields
}

func (*JobRequest) Accept 

func (r *JobRequest) Accept(args ...JobAcceptArguments) error

func (*JobRequest) AgentName added in v0.0.15 

func (r *JobRequest) AgentName() string

func (*JobRequest) ID added in v0.0.15 

func (r *JobRequest) ID() string

func (*JobRequest) Publisher added in v0.0.15 

func (r *JobRequest) Publisher() *livekit.ParticipantInfo

func (*JobRequest) Reject 

func (r *JobRequest) Reject(args ...JobRejectArguments) error

func (*JobRequest) Room added in v0.0.15 

func (r *JobRequest) Room() *livekit.Room

type JobRoomServiceAPI added in v0.0.15 

type JobRoomServiceAPI interface {
	DeleteRoom(context.Context, *livekit.DeleteRoomRequest) (*livekit.DeleteRoomResponse, error)
	MoveParticipant(context.Context, *livekit.MoveParticipantRequest) (*livekit.MoveParticipantResponse, error)
}

type JobSIPAPI added in v0.0.15 

type JobSIPAPI interface {
	CreateSIPParticipant(context.Context, *livekit.CreateSIPParticipantRequest) (*livekit.SIPParticipantInfo, error)
	TransferSIPParticipant(context.Context, *livekit.TransferSIPParticipantRequest) (*emptypb.Empty, error)
}

type LocalJobOptions added in v0.0.15 

type LocalJobOptions struct {
	FakeJob           bool
	RoomInfo          *livekit.Room
	Token             string
	RecordingOptions  agent.RecordingOptions
	SessionReportPath string
	SessionDirectory  string
}

type ParticipantEntrypoint added in v0.0.15 

type ParticipantEntrypoint func(*JobContext, *livekit.ParticipantInfo)

type PlaybackFinishedEvent added in v0.0.15 

type PlaybackFinishedEvent struct {
	PlaybackPosition       time.Duration
	Interrupted            bool
	SynchronizedTranscript string
	AudioFrames            int
	AudioBytes             int
	AudioEncodedFrames     int
	AudioSampleRate        uint32
	AudioChannels          uint32
	AudioLastError         string
}

type PlaybackStartedEvent added in v0.0.15 

type PlaybackStartedEvent struct {
	CreatedAt time.Time
}

type PreConnectAudioBuffer 

type PreConnectAudioBuffer struct {
	Timestamp time.Time
	Frames    []*model.AudioFrame
}

type PreConnectAudioHandler 

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

func NewPreConnectAudioHandler 

func NewPreConnectAudioHandler(room *lksdk.Room, timeout time.Duration) *PreConnectAudioHandler

func (*PreConnectAudioHandler) Close added in v0.0.5 

func (h *PreConnectAudioHandler) Close()

func (*PreConnectAudioHandler) Register 

func (h *PreConnectAudioHandler) Register()

func (*PreConnectAudioHandler) WaitForData 

func (h *PreConnectAudioHandler) WaitForData(ctx context.Context, trackID string) []*model.AudioFrame

type RecorderIO 

type RecorderIO struct {
	Session *agent.AgentSession

	InputStartTime  *time.Time
	OutputStartTime *time.Time
	// contains filtered or unexported fields
}

func NewRecorderIO 

func NewRecorderIO(session *agent.AgentSession) *RecorderIO

func (*RecorderIO) OutputPath added in v0.0.11 

func (r *RecorderIO) OutputPath() string

func (*RecorderIO) PopulateSessionReport added in v0.0.15 

func (r *RecorderIO) PopulateSessionReport(report *agent.SessionReport)

func (*RecorderIO) RecordInput 

func (r *RecorderIO) RecordInput(frame *model.AudioFrame)

func (*RecorderIO) RecordOutput 

func (r *RecorderIO) RecordOutput(frame *model.AudioFrame)

func (*RecorderIO) Recording added in v0.0.15 

func (r *RecorderIO) Recording() bool

func (*RecorderIO) RecordingStartedAt added in v0.0.11 

func (r *RecorderIO) RecordingStartedAt() *time.Time

func (*RecorderIO) Start 

func (r *RecorderIO) Start(outputPath string, sampleRate int) error

func (*RecorderIO) Stop 

func (r *RecorderIO) Stop() error

type RoomIO 

type RoomIO struct {
	Room         *lksdk.Room
	AgentSession *agent.AgentSession
	Options      RoomOptions
	Recorder     *RecorderIO
	// contains filtered or unexported fields
}

func NewRoomIO 

func NewRoomIO(room *lksdk.Room, session *agent.AgentSession, opts RoomOptions) *RoomIO

func (*RoomIO) AttachRoom added in v0.0.15 

func (rio *RoomIO) AttachRoom(room *lksdk.Room)

func (*RoomIO) AudioOutputDiagnostics added in v0.0.51 

func (rio *RoomIO) AudioOutputDiagnostics() RoomIOAudioOutputDiagnostics

func (*RoomIO) ClearBuffer added in v0.0.5 

func (rio *RoomIO) ClearBuffer()

func (*RoomIO) Close 

func (rio *RoomIO) Close() error

func (*RoomIO) Flush added in v0.0.5 

func (rio *RoomIO) Flush()

func (*RoomIO) GetCallback 

func (rio *RoomIO) GetCallback() *lksdk.RoomCallback

func (*RoomIO) LinkedParticipant added in v0.0.15 

func (rio *RoomIO) LinkedParticipant() (string, bool)

func (*RoomIO) OffPlaybackFinished added in v0.0.15 

func (rio *RoomIO) OffPlaybackFinished(callback func(PlaybackFinishedEvent))

func (*RoomIO) OffPlaybackStarted added in v0.0.15 

func (rio *RoomIO) OffPlaybackStarted(callback func(PlaybackStartedEvent))

func (*RoomIO) OnPlaybackFinished added in v0.0.5 

func (rio *RoomIO) OnPlaybackFinished(callback func(PlaybackFinishedEvent))

func (*RoomIO) OnPlaybackStarted added in v0.0.5 

func (rio *RoomIO) OnPlaybackStarted(callback func(PlaybackStartedEvent))

func (*RoomIO) PublishAudio 

func (rio *RoomIO) PublishAudio(frame *model.AudioFrame) error

func (*RoomIO) PublishDTMF added in v0.0.15 

func (rio *RoomIO) PublishDTMF(code int32, digit string) error

func (*RoomIO) SetParticipant added in v0.0.5 

func (rio *RoomIO) SetParticipant(participantIdentity string)

func (*RoomIO) Start 

func (rio *RoomIO) Start(ctx context.Context) error

func (*RoomIO) UnsetParticipant added in v0.0.5 

func (rio *RoomIO) UnsetParticipant()

func (*RoomIO) WaitForPlayout added in v0.0.5 

func (rio *RoomIO) WaitForPlayout(ctx context.Context) (PlaybackFinishedEvent, error)

type RoomIOAudioOutputDiagnostics added in v0.0.51 

type RoomIOAudioOutputDiagnostics struct {
	TrackID                     string
	TrackPublished              bool
	TrackSubscribed             bool
	FramesReceived              int
	FramesPublished             int
	BytesReceived               int
	BytesPublished              int
	EncodedFramesPublished      int
	LastInputSampleRate         uint32
	LastInputSamplesPerChannel  uint32
	LastInputChannels           uint32
	LastPublishedSampleRate     uint32
	LastPublishedSamplesPerChan uint32
	LastPublishedChannels       uint32
	LastFrameAt                 time.Time
	LastPublishedAt             time.Time
	LastError                   string
	LastErrorAt                 time.Time
}

type RoomOptions 

type RoomOptions struct {
	AudioTrackName             string
	AudioSubscriptionTimeout   time.Duration
	PreConnectAudioTimeout     time.Duration
	DisablePreConnectAudio     bool
	DisableAudioInput          bool
	DisableTextInput           bool
	DisableAudioOutput         bool
	DisableTranscriptionOutput bool
	DisableCloseOnDisconnect   bool
	DeleteRoomOnClose          bool
	DeleteRoom                 func(context.Context, string) error
	TextInputCallback          TextInputCallback
	ParticipantIdentity        string
	ParticipantKinds           []lksdk.ParticipantKind
}

type TextInputCallback added in v0.0.15 

type TextInputCallback func(context.Context, *agent.AgentSession, TextInputEvent) error

type TextInputEvent added in v0.0.15 

type TextInputEvent struct {
	Text                string
	Info                lksdk.TextStreamInfo
	ParticipantIdentity string
}

type WorkerInfo added in v0.0.15 

type WorkerInfo struct {
	HTTPPort    int
	CloudAgents bool
}

type WorkerOptions 

type WorkerOptions struct {
	AgentName      string
	AgentNameIsEnv bool
	WorkerType     WorkerType
	MaxRetry       int
	MaxRetrySet    bool
	Version        string
	Host           string
	Port           int
	PortSet        bool
	WSURL          string
	LoadFunc       func(*AgentServer) float64
	HealthCheck    func(*AgentServer) error
	SetupFunc      func(*JobProcess) error
	// WSRL is kept for backward compatibility. Prefer WSURL for new code.
	WSRL                               string
	APIKey                             string
	APISecret                          string
	WorkerToken                        string
	HTTPProxy                          string
	HTTPProxySet                       bool
	UserArguments                      any
	DevMode                            bool
	LogLevel                           string
	PrometheusPort                     int
	PrometheusPortSet                  bool
	PrometheusMultiprocDir             string
	LoadThreshold                      float64
	LoadThresholdSet                   bool
	JobMemoryWarnMB                    float64
	JobMemoryWarnMBSet                 bool
	JobMemoryLimitMB                   float64
	JobMemoryLimitMBSet                bool
	NumIdleProcesses                   int
	NumIdleProcessesSet                bool
	DrainTimeoutSeconds                int
	DrainTimeoutSecondsSet             bool
	SessionEndTimeoutSeconds           float64
	SessionEndTimeoutSecondsSet        bool
	ShutdownProcessTimeoutSeconds      float64
	ShutdownProcessTimeoutSecondsSet   bool
	InitializeProcessTimeoutSeconds    float64
	InitializeProcessTimeoutSecondsSet bool
	Permissions                        *WorkerPermissions
}

type WorkerPermissions added in v0.0.15 

type WorkerPermissions struct {
	CanPublish        bool
	CanSubscribe      bool
	CanPublishData    bool
	CanUpdateMetadata bool
	CanPublishSources []livekit.TrackSource
	Hidden            bool
}

type WorkerRegisteredHandler added in v0.0.15 

type WorkerRegisteredHandler func(workerID string, serverInfo *livekit.ServerInfo)

type WorkerStartedHandler added in v0.0.15 

type WorkerStartedHandler func()

type WorkerType 

type WorkerType string
const (
	WorkerTypeRoom      WorkerType = "room"
	WorkerTypePublisher WorkerType = "publisher"
)

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.