workers

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: BSD-2-Clause Imports: 18 Imported by: 0

Documentation

Overview

Package workers is the base every worker is built on: the thing that connects to a bus, registers itself so others can find it, is activated and deactivated, and exchanges long-running jobs with the other workers.

A worker is a participant in a session rather than a step in a pipeline. One may drive a pipeline of its own, another may only listen on the bus, and a parent may add children whose lifetime it manages. What they share is the protocol in this package: become ready, be told to start or stop, ask each other for work and report back on it.

A worker embeds *Base and hands itself to New, which is what lets the base reach the worker's own versions of the hooks it calls: 

type Greeter struct{ *workers.Base }

func NewGreeter() *Greeter {
    g := &Greeter{}
    g.Base = workers.New(workers.Config{Name: "greeter"}, g)
    return g
}

func (g *Greeter) OnJobRequest(ctx context.Context, m *bus.JobRequestMessage) {
    _ = g.SendJobResponse(ctx, m.JobID, map[string]any{"hello": true}, workers.JobResponseOptions{})
}

A worker that overrides a hook the base calls should call the base's version too, the way Greeter would call g.Base.OnJobRequest, unless it means to replace the behavior entirely.

Index

Constants

View Source 
const (
	// EventRunnerReady fires once the runner has set itself up and started the
	// workers added to it.
	EventRunnerReady = "on_ready"
	// EventRunnerError fires when a worker added to the runner fails to start.
	EventRunnerError = "on_error"
)

The events a runner raises.

View Source 
const (
	// EventActivated fires when this worker is activated, with the arguments it
	// was activated with.
	EventActivated = "on_activated"
	// EventDeactivated fires when this worker is deactivated.
	EventDeactivated = "on_deactivated"
	// EventBusMessage fires for every bus message, after the built-in handling
	// of it.
	EventBusMessage = "on_bus_message"
	// EventJobRequest fires when this worker is asked to do a job.
	EventJobRequest = "on_job_request"
	// EventJobResponse fires when a worker reports how its job ended.
	EventJobResponse = "on_job_response"
	// EventJobUpdate fires when a worker reports progress on a job.
	EventJobUpdate = "on_job_update"
	// EventJobUpdateRequested fires when the requester asks for progress on a
	// job this worker is doing.
	EventJobUpdateRequested = "on_job_update_requested"
	// EventJobCompleted fires when every worker in a group has responded.
	EventJobCompleted = "on_job_completed"
	// EventJobError fires when a worker failed and the group was called off.
	EventJobError = "on_job_error"
	// EventJobStreamStart fires when a worker opens a stream of results.
	EventJobStreamStart = "on_job_stream_start"
	// EventJobStreamData fires for each item of a worker's result stream.
	EventJobStreamData = "on_job_stream_data"
	// EventJobStreamEnd fires when a worker closes its result stream.
	EventJobStreamEnd = "on_job_stream_end"
	// EventJobCancelled fires when a job this worker is doing is called off.
	EventJobCancelled = "on_job_cancelled" //nolint:misspell // the event name the protocol uses
	// EventWorkerReady fires when another worker becomes ready.
	EventWorkerReady = "on_worker_ready"
	// EventWorkerFailed fires when a child worker reports an error.
	EventWorkerFailed = "on_worker_failed"
)

The events a worker raises. Attach handlers with the Add method it inherits from the event registry.

Variables

View Source 
var ErrNoActiveJob = errors.New("no active job of that id")

ErrNoActiveJob is reported when a worker answers or reports on a job it is not working on, usually one it has already finished or that was called off.

View Source 
var ErrNoRegistry = errors.New("worker is not attached to a registry")

ErrNoRegistry is reported when a worker asks for a job before it has been attached to a registry, so it cannot tell whether the workers exist.

Functions

This section is empty.

Types

type ActivateOptions 

type ActivateOptions struct {
	// Args are handed to the target worker's OnActivated, and may be nil.
	Args ActivationArgs
	// DeactivateSelf deactivates this worker before activating the target, so
	// the two are never briefly active together.
	DeactivateSelf bool
}

ActivateOptions are the choices activating another worker offers.

type ActivationArgs 

type ActivationArgs interface {
	// ToMap is the arguments as they travel on the bus, leaving out what is
	// unset.
	ToMap() map[string]any
}

ActivationArgs is what a worker is handed when it is activated. A worker with arguments of its own defines a type for them, embeds BaseActivationArgs and puts its own fields in the map alongside.

type Base 

type Base struct {
	events.Registry
	// contains filtered or unexported fields
}

Base is the worker every worker is built on. See the package documentation for how to build one.

It is safe for concurrent use: the bus hands a worker its system messages and its data messages on separate goroutines, so a cancel arrives while the work it is calling off is still being handled.

func New 

func New(cfg Config, self Worker) *Base

New builds a worker base. self is the worker embedding it, which is how the base reaches that worker's own hooks; see the package documentation.

func (*Base) AcceptsBusMessage 

func (w *Base) AcceptsBusMessage(m bus.Message) bool

AcceptsBusMessage takes bus messages only while the worker is active.

An inactive worker is handed only activation, deactivation, end or cancel. Work addressed to it, and everything it would merely observe, is dropped by the bus rather than reaching OnBusMessage. Registry notifications are unaffected, since a ready handler fires from the worker registry rather than traveling over the bus.

func (*Base) ActivateWorker 

func (w *Base) ActivateWorker(ctx context.Context, workerName string, opts ActivateOptions)

ActivateWorker activates another worker by name, which calls that worker's OnActivated with the arguments given.

func (*Base) ActivationArgs 

func (w *Base) ActivationArgs() map[string]any

ActivationArgs are the arguments of the most recent activation, and are nil while the worker is inactive.

func (*Base) Active 

func (w *Base) Active() bool

Active reports whether the worker is accepting bus messages.

An active worker takes everything addressed to it. An inactive one takes only activation, deactivation, end or cancel, so no job request, frame or UI event reaches it and none of its message handling runs.

It matters mainly in a multi-worker setup, where a worker is put out of the way while the others carry on. Registry watches sit outside it: a ready handler fires whatever this reports, because it never travels over the bus.

func (*Base) ActiveJobs 

func (w *Base) ActiveJobs() map[string]*bus.JobRequestMessage

ActiveJobs are the job requests this worker is working on, by job id.

func (*Base) AddWorkers 

func (w *Base) AddWorkers(ctx context.Context, children ...Worker)

AddWorkers puts workers under this one as its children, and watches each, so OnWorkerReady fires as each becomes ready.

A parent manages its children's lifetime: ending or canceling it ends or cancels them. A worker that already has a parent is left where it is.

func (*Base) AddWorkersUnwatched 

func (w *Base) AddWorkersUnwatched(ctx context.Context, children ...Worker)

AddWorkersUnwatched is AddWorkers without watching the children, for a parent that does not need to be told when they become ready. It can still call WatchWorkers later.

func (*Base) Attach 

func (w *Base) Attach(_ context.Context, reg *registry.WorkerRegistry, b *bus.Bus)

Attach connects the worker to the registry and bus its runner provides, and subscribes it. It is called before the worker runs, so a worker added later is listening before any of them sends its first message.

func (*Base) Bridged 

func (w *Base) Bridged() bool

Bridged reports whether the worker is bridged onto the bus. A worker that wraps its pipeline in bus edges overrides it.

func (*Base) Bus 

func (w *Base) Bus() *bus.Bus

Bus is the bus the worker is attached to, and is nil until Attach.

func (*Base) Cancel 

func (w *Base) Cancel(ctx context.Context, reason string)

Cancel asks for every worker to stop at once.

func (*Base) CancelGroup 

func (w *Base) CancelGroup(ctx context.Context, jobID, reason string)

CancelGroup calls a running job group off, telling each of its workers.

func (*Base) Children 

func (w *Base) Children() []Worker

Children are the workers this one added.

func (*Base) Cleanup 

func (w *Base) Cleanup(ctx context.Context)

Cleanup releases the worker and stops it. It waits for the handlers of the worker's own events before stopping, so none is left running past the worker it was watching.

func (*Base) CreateGroupAndRequestJob 

func (w *Base) CreateGroupAndRequestJob(
	ctx context.Context, workerNames []string, req jobcontext.Request,
) (*jobcontext.Group, error)

CreateGroupAndRequestJob waits for the named workers to be ready, opens a group for them and sends each the request. It does not wait for the group; use Group.Wait, or JobGroup, for that.

It reports ErrGroup when the workers are not all ready within the request's timeout.

func (*Base) DeactivateWorker 

func (w *Base) DeactivateWorker(ctx context.Context, workerName string)

DeactivateWorker deactivates another worker by name, which calls that worker's OnDeactivated.

func (*Base) End 

func (w *Base) End(ctx context.Context, reason string)

End asks for the session to end gracefully.

func (*Base) Events 

func (w *Base) Events() *events.Registry

Events is the worker's event registry, which is where a typed handler is attached with events.On: 

events.On(worker.Events(), pipeline.EventPipelineFinished, handler)

func (*Base) Finished 

func (w *Base) Finished() <-chan struct{}

Finished is closed when the worker finishes.

func (*Base) HandleJob 

func (w *Base) HandleJob(name string, opts JobOptions, fn JobHandler)

HandleJob declares fn as this worker's handler for job requests named name. A request naming no job, or one no handler was declared for, goes to OnJobRequest instead.

Declare handlers when building the worker. Declaring two for one name panics: the second would never run, and a worker whose handlers are not the ones its author declared cannot do the work it was built for. It is a mistake in the program rather than something that goes wrong at run time, so it is refused where it is made.

func (*Base) HandleWorkerCancel 

func (w *Base) HandleWorkerCancel(ctx context.Context, m *bus.CancelWorkerMessage)

HandleWorkerCancel passes the cancel on to the children and stops. See HandleWorkerEnd.

func (*Base) HandleWorkerEnd 

func (w *Base) HandleWorkerEnd(ctx context.Context, m *bus.EndWorkerMessage)

HandleWorkerEnd passes the end on to the children, waits for them, and then stops. A worker with a runtime of its own overrides it, calls PropagateEndToChildren and drives its own shutdown, so that it finishes at the right moment.

func (*Base) HandleWorkerReady 

func (w *Base) HandleWorkerReady(name string, fn ReadyHandler)

HandleWorkerReady declares fn as this worker's handler for the named worker becoming ready. The worker is watched when this one starts, and fn runs before the general OnWorkerReady hook.

Declaring two handlers for one worker panics, for the reason HandleJob does.

func (*Base) HasGroup 

func (w *Base) HasGroup(jobID string) bool

HasGroup reports whether a group of that id is still running.

func (*Base) Job 

func (w *Base) Job(
	ctx context.Context, workerName string, req jobcontext.Request, block func(j *jobcontext.Job) error,
) (*jobcontext.Job, error)

Job asks one worker for a job, runs block while it works, and waits for it on the way out. See jobcontext.RunJob.

func (*Base) JobGroup 

func (w *Base) JobGroup(
	ctx context.Context, workerNames []string, req jobcontext.Request, block func(g *jobcontext.Group) error,
) (*jobcontext.Group, error)

JobGroup asks several workers for a job together, runs block while they work, and waits for them all on the way out. See jobcontext.RunGroup.

func (*Base) JobGroups 

func (w *Base) JobGroups() map[string]*jobcontext.Group

JobGroups are the job groups this worker launched and is still waiting on, by job id.

func (*Base) Name 

func (w *Base) Name() string

Name is what other workers address this one by, and identifies it on the bus.

func (*Base) OnActivated 

func (w *Base) OnActivated(context.Context, map[string]any)

OnActivated is called when this worker is activated. The default does nothing.

func (*Base) OnBusMessage 

func (w *Base) OnBusMessage(ctx context.Context, m bus.Message)

OnBusMessage handles one bus message: the built-in lifecycle and job handling, and then the on_bus_message event.

A worker with message types of its own overrides it, calls this one, and handles the rest itself.

func (*Base) OnDeactivated 

func (w *Base) OnDeactivated(context.Context)

OnDeactivated is called when this worker is deactivated. The default does nothing.

func (*Base) OnJobCancelled 

func (w *Base) OnJobCancelled(context.Context, *bus.JobCancelMessage)

OnJobCancelled is called when a job this worker is doing is called off. Override it to release what the work was holding; the canceled answer is sent for you afterwards.

func (*Base) OnJobCompleted 

func (w *Base) OnJobCompleted(context.Context, jobcontext.GroupResponse)

OnJobCompleted is called when every worker in a group has answered.

func (*Base) OnJobError 

func (w *Base) OnJobError(context.Context, bus.JobResponse)

OnJobError is called when a worker reported it failed and the group was called off, so OnJobCompleted will not fire. What the other workers answered before it is in the group's responses.

func (*Base) OnJobRequest 

func (w *Base) OnJobRequest(context.Context, *bus.JobRequestMessage)

OnJobRequest is called when this worker is asked for a job no declared handler matched. Override it to do the work. The default does nothing, which leaves the requester waiting.

func (*Base) OnJobResponse 

func (w *Base) OnJobResponse(context.Context, bus.JobResponse)

OnJobResponse is called when a worker reports how its job ended. Override it to take each answer as it arrives, rather than waiting for the group.

func (*Base) OnJobStreamData 

func (w *Base) OnJobStreamData(context.Context, *bus.JobStreamDataMessage)

OnJobStreamData is called for each item of a worker's result stream.

func (*Base) OnJobStreamEnd 

func (w *Base) OnJobStreamEnd(context.Context, *bus.JobStreamEndMessage)

OnJobStreamEnd is called when a worker closes its result stream.

func (*Base) OnJobStreamStart 

func (w *Base) OnJobStreamStart(context.Context, *bus.JobStreamStartMessage)

OnJobStreamStart is called when a worker opens a stream of results.

func (*Base) OnJobUpdate 

func (w *Base) OnJobUpdate(context.Context, bus.JobUpdate)

OnJobUpdate is called when a worker reports progress on a job.

func (*Base) OnJobUpdateRequested 

func (w *Base) OnJobUpdateRequested(context.Context, *bus.JobUpdateRequestMessage)

OnJobUpdateRequested is called when the requester asks how far along this worker is. Override it to answer with SendJobUpdate.

func (*Base) OnWorkerFailed 

func (w *Base) OnWorkerFailed(context.Context, registry.WorkerErrorData)

OnWorkerFailed is called when a child worker reports an error. The default does nothing.

func (*Base) OnWorkerReady 

func (w *Base) OnWorkerReady(context.Context, registry.WorkerReadyData)

OnWorkerReady is called when another worker becomes ready.

It fires for a local root worker on its own, for a child only on the parent that added it, and for a remote worker only when it is watched. The default does nothing.

func (*Base) Parent 

func (w *Base) Parent() string

Parent is the name of the worker that added this one, empty for a root worker.

func (*Base) PropagateCancelToChildren 

func (w *Base) PropagateCancelToChildren(ctx context.Context, reason string)

PropagateCancelToChildren asks each child to stop at once.

func (*Base) PropagateEndToChildren 

func (w *Base) PropagateEndToChildren(ctx context.Context, reason string)

PropagateEndToChildren asks each child to end and waits for them all.

func (*Base) RequestJob 

func (w *Base) RequestJob(ctx context.Context, workerName string, req jobcontext.Request) (string, error)

RequestJob asks one worker for a job and does not wait for it.

It waits for the worker to be ready, sends the request and returns the job's id. Watch for the answer with OnJobResponse or OnJobCompleted, or use Job to wait for it.

func (*Base) RequestJobGroup 

func (w *Base) RequestJobGroup(ctx context.Context, workerNames []string, req jobcontext.Request) (string, error)

RequestJobGroup asks several workers for a job together and does not wait for it. It returns the id they all share.

func (*Base) RequestJobUpdate 

func (w *Base) RequestJobUpdate(ctx context.Context, jobID, workerName string)

RequestJobUpdate asks a worker how far along it is with a job.

func (*Base) Run 

func (w *Base) Run(ctx context.Context) error

Run runs the worker until it finishes.

This is the plain bus-only worker: it starts, then waits to be stopped. A worker with a runtime of its own overrides it.

func (*Base) SendBusErrorMessage 

func (w *Base) SendBusErrorMessage(ctx context.Context, workerErr string)

SendBusErrorMessage reports that this worker failed.

A child reports to its parent, in this process only. A root worker reports to everyone, over the network.

func (*Base) SendBusMessage 

func (w *Base) SendBusMessage(ctx context.Context, m bus.Message)

SendBusMessage puts a message on the bus. It is a no-op for a worker that is not attached to one.

func (*Base) SendJobResponse 

func (w *Base) SendJobResponse(
	ctx context.Context, jobID string, response map[string]any, opts JobResponseOptions,
) error

SendJobResponse answers the worker that asked for a job, and ends this worker's part in it.

func (*Base) SendJobStreamData 

func (w *Base) SendJobStreamData(ctx context.Context, jobID string, data map[string]any) error

SendJobStreamData sends one item of a job's result stream.

func (*Base) SendJobStreamEnd 

func (w *Base) SendJobStreamEnd(ctx context.Context, jobID string, data map[string]any) error

SendJobStreamEnd closes a job's result stream, and ends this worker's part in the job: the stream's end is its answer, so a cancel arriving afterwards finds nothing to call off.

func (*Base) SendJobStreamStart 

func (w *Base) SendJobStreamStart(ctx context.Context, jobID string, data map[string]any) error

SendJobStreamStart opens a stream of results back to the worker that asked for the job.

func (*Base) SendJobUpdate 

func (w *Base) SendJobUpdate(ctx context.Context, jobID string, update map[string]any, urgent bool) error

SendJobUpdate reports progress on a job to the worker that asked for it. Urgent delivers it ahead of the data messages already queued.

func (*Base) Start 

func (w *Base) Start(ctx context.Context)

Start marks the worker started, registers it as ready, activates it if it was built active, and watches the workers it declared ready handlers for.

func (*Base) StartedAt 

func (w *Base) StartedAt() float64

StartedAt is when the worker became ready, as a Unix timestamp, and is zero until it has.

func (*Base) Stop 

func (w *Base) Stop(ctx context.Context)

Stop calls off everything the worker had running and marks it finished.

Every job group it launched is called off, and every job request it was still working on is answered as canceled, so nobody is left waiting on a worker that has stopped.

func (*Base) Wait 

func (w *Base) Wait(ctx context.Context)

Wait blocks until the worker finishes, or ctx ends.

func (*Base) WatchWorkers 

func (w *Base) WatchWorkers(ctx context.Context, workerNames ...string)

WatchWorkers asks to be told when the named workers register. A worker that has already registered is reported straight away.

func (*Base) WorkerRegistry 

func (w *Base) WorkerRegistry() *registry.WorkerRegistry

WorkerRegistry is the shared worker registry, and is nil until Attach.

type BaseActivationArgs 

type BaseActivationArgs struct {
	// Metadata is structured data for the worker being activated, and may be
	// nil.
	Metadata map[string]any
}

BaseActivationArgs is the part of the activation arguments every worker understands.

func BaseActivationArgsFrom 

func BaseActivationArgsFrom(args map[string]any) BaseActivationArgs

BaseActivationArgsFrom reads the part every worker understands out of the arguments a worker was activated with, ignoring anything else in them.

func (BaseActivationArgs) ToMap 

func (a BaseActivationArgs) ToMap() map[string]any

ToMap implements ActivationArgs.

type Config 

type Config struct {
	// Name is what other workers address this one by, and must be unique among
	// them. Empty names it after its type, which suits a worker taking no part
	// in worker-to-worker messaging.
	Name string
	// Active reports whether the worker starts active; nil defaults to true.
	Active *bool
}

Config configures a worker.

type JobHandler 

type JobHandler func(ctx context.Context, m *bus.JobRequestMessage)

JobHandler does the work one kind of job asks for. Report progress with SendJobUpdate and finish with SendJobResponse; a handler that returns without answering leaves the requester waiting until it gives up.

type JobOptions 

type JobOptions struct {
	// Sequential runs the requests for this job one at a time, in the order
	// they arrived, rather than concurrently. Waiting counts against the
	// requester's timeout, so a slow predecessor can time a queued request out
	// before it starts.
	Sequential bool
}

JobOptions are the choices a job handler is declared with.

type JobResponseOptions 

type JobResponseOptions struct {
	// Status is how the job ended; empty reports it completed.
	Status jobcontext.JobStatus
	// Urgent delivers the answer ahead of the data messages already queued.
	Urgent bool
}

JobResponseOptions are the choices answering a job offers.

type ReadyHandler 

type ReadyHandler func(ctx context.Context, data registry.WorkerReadyData)

ReadyHandler is called when the worker it was declared for becomes ready.

type RunOptions 

type RunOptions struct {
	// AutoEnd ends the runner once every root worker has finished, which is
	// what makes a single-pipeline bot end when its pipeline does; nil defaults
	// to true. Set it false for a host that adds and removes workers across
	// many sessions, which should outlive any of them.
	AutoEnd *bool
}

RunOptions are the choices running offers.

type Runner 

type Runner struct {
	events.Registry
	// contains filtered or unexported fields
}

Runner runs workers to completion. It owns the bus they talk over, the registry they find each other through, and the goroutines they run on.

Add the workers with AddWorkers and then call Run: 

runner := workers.NewRunner(workers.RunnerConfig{})
runner.AddWorkers(ctx, worker)
err := runner.Run(ctx, workers.RunOptions{})

Run ends once every root worker has finished, so a bot with one pipeline ends when that pipeline does. A bot whose helpers wait on the bus forever ends by calling End or Cancel instead.

func NewRunner 

func NewRunner(cfg RunnerConfig) *Runner

NewRunner builds a runner.

func (*Runner) AddWorkers 

func (r *Runner) AddWorkers(ctx context.Context, added ...Worker)

AddWorkers registers workers on the runner, attaching each to the bus and registry and starting it.

A worker added before Run is started when the runner sets itself up; one added while it is running starts there and then. A name already registered is reported and skipped.

func (*Runner) Bus 

func (r *Runner) Bus() *bus.Bus

Bus is the bus this runner hosts and shares with its workers.

func (*Runner) Cancel 

func (r *Runner) Cancel(_ context.Context, reason string)

Cancel stops every worker at once. It records why and signals the shutdown, which the run answers by canceling each worker still going and waiting for it. Calling it again does nothing.

The messages go out from that one exit path rather than from here, so a worker is told once rather than twice on an ordinary shutdown, and the caller's reason travels with them.

func (*Runner) End 

func (r *Runner) End(ctx context.Context, reason string)

End asks every worker that has not finished to stop gracefully. Calling it again does nothing.

func (*Runner) Name 

func (r *Runner) Name() string

Name identifies the runner on the bus.

func (*Runner) OnBusMessage 

func (r *Runner) OnBusMessage(ctx context.Context, m bus.Message)

OnBusMessage handles the messages that are the runner's business rather than any one worker's.

func (*Runner) Run 

func (r *Runner) Run(ctx context.Context, opts RunOptions) error

Run starts the workers added to the runner and blocks until it stops.

func (*Runner) StopWhenDone 

func (r *Runner) StopWhenDone()

StopWhenDone asks every root worker that can to finish what it is doing and then stop.

func (*Runner) String 

func (r *Runner) String() string

String names the runner in a message.

func (*Runner) WorkerRegistry 

func (r *Runner) WorkerRegistry() *registry.WorkerRegistry

WorkerRegistry is the registry this runner owns and shares with its workers.

type RunnerConfig 

type RunnerConfig struct {
	// Name identifies the runner, and must be unique among the runners sharing
	// a bus. Empty generates one.
	Name string
	// Bus is the bus to host and share with the workers; nil builds an
	// in-process one.
	Bus *bus.Bus
	// HandleInterrupt cancels the runner on an interrupt signal; nil defaults
	// to true.
	HandleInterrupt *bool
	// HandleTerminate cancels the runner on a termination signal.
	HandleTerminate bool
	// ForceGC collects garbage once every worker has finished.
	ForceGC bool
}

RunnerConfig configures a Runner.

type StoppableWhenDone 

type StoppableWhenDone interface {
	Worker
	// StopWhenDone stops the worker once it has finished what it is doing.
	StopWhenDone()
}

StoppableWhenDone is a worker that can be asked to finish what it is doing and then stop, rather than being stopped where it stands. A worker driving a pipeline is one; a worker that only listens on the bus is not.

type Worker 

type Worker interface {
	bus.Subscriber

	// Bridged reports whether the worker is bridged onto the bus, which is
	// announced when it becomes ready.
	Bridged() bool
	// Attach connects the worker to the registry and bus its runner provides.
	Attach(ctx context.Context, reg *registry.WorkerRegistry, b *bus.Bus)
	// Run runs the worker until it finishes.
	Run(ctx context.Context) error
	// Stop cleans up and marks the worker finished.
	Stop(ctx context.Context)
	// Wait blocks until the worker finishes, or ctx ends.
	Wait(ctx context.Context)
	// Cleanup releases the worker and stops it.
	Cleanup(ctx context.Context)

	// OnActivated is called when this worker is activated.
	OnActivated(ctx context.Context, args map[string]any)
	// OnDeactivated is called when this worker is deactivated.
	OnDeactivated(ctx context.Context)
	// OnWorkerReady is called when another worker becomes ready.
	OnWorkerReady(ctx context.Context, data registry.WorkerReadyData)
	// OnWorkerFailed is called when a child worker reports an error.
	OnWorkerFailed(ctx context.Context, data registry.WorkerErrorData)
	// OnJobRequest is called when this worker is asked to do a job.
	OnJobRequest(ctx context.Context, m *bus.JobRequestMessage)
	// OnJobResponse is called when a worker reports how its job ended.
	OnJobResponse(ctx context.Context, m bus.JobResponse)
	// OnJobUpdate is called when a worker reports progress on a job.
	OnJobUpdate(ctx context.Context, m bus.JobUpdate)
	// OnJobUpdateRequested is called when the requester asks for progress.
	OnJobUpdateRequested(ctx context.Context, m *bus.JobUpdateRequestMessage)
	// OnJobCompleted is called when every worker in a group has responded.
	OnJobCompleted(ctx context.Context, result jobcontext.GroupResponse)
	// OnJobError is called when a worker failed and the group was called off.
	OnJobError(ctx context.Context, m bus.JobResponse)
	// OnJobStreamStart is called when a worker opens a stream of results.
	OnJobStreamStart(ctx context.Context, m *bus.JobStreamStartMessage)
	// OnJobStreamData is called for each item of a worker's result stream.
	OnJobStreamData(ctx context.Context, m *bus.JobStreamDataMessage)
	// OnJobStreamEnd is called when a worker closes its result stream.
	OnJobStreamEnd(ctx context.Context, m *bus.JobStreamEndMessage)
	// OnJobCancelled is called when a job this worker is doing is called off.
	OnJobCancelled(ctx context.Context, m *bus.JobCancelMessage)

	// HandleWorkerEnd is called when this worker is asked to end gracefully.
	// The default passes the end to the children and stops; a worker with a
	// runtime of its own drives that runtime's shutdown instead, so that it
	// finishes at the right moment.
	HandleWorkerEnd(ctx context.Context, m *bus.EndWorkerMessage)
	// HandleWorkerCancel is called when this worker is asked to stop at once.
	// The default passes the cancel to the children and stops; see
	// HandleWorkerEnd.
	HandleWorkerCancel(ctx context.Context, m *bus.CancelWorkerMessage)
	// contains filtered or unexported methods
}

Worker is a worker as the base sees it: the hooks the base calls on the worker it belongs to, so a worker's own version of one is reached rather than the base's.

Only a type embedding *Base satisfies it, which is the point: the hooks all have a default, and a worker overrides the ones it cares about.

Source Files

View all Source files

Directories

Path Synopsis
Package llmworker is a pipeline worker built around a language model, with the tool handling that makes one usable from the bus.
 Package llmworker is a pipeline worker built around a language model, with the tool handling that makes one usable from the bus.
Package proxy carries bus messages between two processes over a WebSocket.
 Package proxy carries bus messages between two processes over a WebSocket.

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.