bus

type ActivateWorkerMessage ¶

type ActivateWorkerMessage struct {
	BaseDataMessage
	// Args are the activation arguments, and may be nil.
	Args map[string]any
}

ActivateWorkerMessage asks a worker to become active.

type AddWorkerMessage ¶

type AddWorkerMessage struct {
	BaseSystemMessage

	// Worker is the worker to manage.
	Worker any
	// contains filtered or unexported fields
}

AddWorkerMessage hands a worker to the runner to manage. It never leaves the process, because it carries the worker itself rather than a description of one.

Worker is typed as any because the workers are built on this package and so cannot be named here. Set it to a worker; anything else is ignored by the runner that receives it.

type AsyncQueueBus ¶

type AsyncQueueBus struct {
	*Bus
}

AsyncQueueBus delivers messages in process, straight to the local subscribers. It is the bus a runner and its workers use when they all live in one process, and the only one the rest of the framework needs.

func NewAsyncQueueBus() *AsyncQueueBus

func (b *AsyncQueueBus) Publish(_ context.Context, m Message)

type BaseDataMessage ¶

type BaseDataMessage struct{ BaseMessage }

BaseDataMessage is embedded by every message delivered in send order.

type BaseMessage ¶

type BaseMessage struct {
	// From is the name of the sender.
	From string
	// To is the name of the intended recipient, or "" to broadcast.
	To string
}

BaseMessage carries what every message has. Embed BaseDataMessage or BaseSystemMessage rather than this directly.

func (m *BaseMessage) Source() string

func (m *BaseMessage) Target() string

type BaseSystemMessage ¶

type BaseSystemMessage struct{ BaseMessage }

BaseSystemMessage is embedded by every message delivered ahead of the queued data messages.

type BridgeConfig ¶

type BridgeConfig struct {
	// Bus is the bus to exchange frames over. Required.
	Bus *Bus
	// WorkerName is the owning worker's name, carried as the source of
	// everything this bridge sends.
	WorkerName string
	// TargetWorker, when set, is the only worker this bridge accepts frames
	// from.
	TargetWorker string
	// Bridge, when set, names this bridge for routing: what it sends is tagged
	// with the name, and it accepts only frames tagged with the same one.
	Bridge string
	// ExcludeFrames are frame types that never cross the bus, on top of the
	// lifecycle frames, which never do. A frame matches when it is of the same
	// concrete type as one given here.
	ExcludeFrames []frames.Frame
}

BridgeConfig configures a BridgeProcessor.

type BridgeProcessor ¶

type BridgeProcessor struct {
	*processor.Base
	// contains filtered or unexported fields
}

BridgeProcessor exchanges frames with other workers over the bus, from the middle of a pipeline.

A frame reaching it is taken off the local pipeline and sent to the bus; a frame arriving from the bus is pushed on locally. Lifecycle frames, urgent transport messages and any excluded types pass straight through instead.

func NewBridgeProcessor(cfg BridgeConfig) *BridgeProcessor

func (p *BridgeProcessor) Cleanup(ctx context.Context) error

func (p *BridgeProcessor) OnBusMessage(ctx context.Context, m Message)

func (p *BridgeProcessor) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

func (p *BridgeProcessor) Setup(ctx context.Context, s processor.Setup) error

type BridgedWorker ¶

type BridgedWorker interface {
	// Bus is the bus the worker is attached to. It is read when the edge is set
	// up rather than when it is built, so the worker need only be attached by
	// then.
	Bus() *Bus
	// Name is the worker's name, carried as the source of what the edge sends.
	Name() string
	// Active reports whether the worker is active. An inactive worker takes no
	// frames off the bus.
	Active() bool
	// QueueFrame hands a frame to the worker's own queue.
	QueueFrame(f frames.Frame, dir ...processor.Direction)
}

BridgedWorker is the part of a pipeline worker an EdgeProcessor reads. It is named here rather than imported because the workers are built on this package; a pipeline worker satisfies it.

type Bus ¶

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

Bus is the transport-independent core of a worker bus. Embed it in a concrete bus and implement Publisher:

type MyBus struct{ *bus.Bus }

func (b *MyBus) Publish(ctx context.Context, m bus.Message) { … }

The zero value is not usable; build one with New.

func New(self Publisher) *Bus

func (b *Bus) OnMessageReceived(m Message)

func (b *Bus) Send(ctx context.Context, m Message)

func (b *Bus) Start(ctx context.Context)

func (b *Bus) Stop()

func (b *Bus) Subscribe(s Subscriber)

func (b *Bus) Unsubscribe(s Subscriber)

type CancelMessage ¶

type CancelMessage struct {
	BaseSystemMessage
	// Reason describes why, and may be empty.
	Reason string
}

CancelMessage stops everything at once, without flushing queued work.

type CancelWorkerMessage ¶

type CancelWorkerMessage struct {
	BaseSystemMessage
	// Reason describes why, and may be empty.
	Reason string
}

CancelWorkerMessage stops one worker at once.

type DeactivateWorkerMessage ¶

type DeactivateWorkerMessage struct{ BaseDataMessage }

DeactivateWorkerMessage asks a worker to become inactive.

type DeserializeFunc ¶

type DeserializeFunc func(v any) (any, error)

DeserializeFunc reconstructs one value from its wire form, and is the counterpart handed to a TypeAdapter.

type EdgeConfig ¶

type EdgeConfig struct {
	// Worker is the worker that owns the pipeline. Required.
	Worker BridgedWorker
	// Direction is the direction this edge copies to the bus. Frames arriving
	// from the bus traveling the other way are injected here.
	Direction processor.Direction
	// Bridges are the bridge names this edge accepts frames from. Empty accepts
	// every bridge.
	Bridges []string
	// ExcludeFrames are frame types that never cross the bus, on top of the
	// lifecycle frames.
	ExcludeFrames []frames.Frame
}

EdgeConfig configures an EdgeProcessor.

type EdgeProcessor ¶

type EdgeProcessor struct {
	*processor.Base
	// contains filtered or unexported fields
}

EdgeProcessor tees a pipeline's edge onto the bus: frames carry on locally and matching ones are copied across.

It is placed by a pipeline worker at the source and sink of a bridged pipeline. Upstream keeps it private to the worker; Go's package boundary makes that impossible, so it is exported and documented as the worker's own.

func NewEdgeProcessor(cfg EdgeConfig) *EdgeProcessor

func (p *EdgeProcessor) Cleanup(ctx context.Context) error

func (p *EdgeProcessor) OnBusMessage(ctx context.Context, m Message)

func (p *EdgeProcessor) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error

func (p *EdgeProcessor) Setup(ctx context.Context, s processor.Setup) error

type EndMessage ¶

type EndMessage struct {
	BaseDataMessage
	// Reason describes why, and may be empty.
	Reason string
}

EndMessage asks everything to shut down gracefully.

type EndWorkerMessage ¶

type EndWorkerMessage struct {
	BaseDataMessage
	// Reason describes why, and may be empty.
	Reason string
}

EndWorkerMessage asks one worker to shut down gracefully.

type FlushProgressMessage ¶

type FlushProgressMessage struct {
	BaseDataMessage
	// FlushID is the id of the probe being reported on.
	FlushID uint64
}

FlushProgressMessage reports that a flush probe from another worker is still making progress. A probe that crosses into another pipeline is answered there, so the worker waiting on it cannot see whether anything is happening. The pipeline holding the probe says so, and the wait stays alive for as long as it keeps saying it.

type FrameMessage ¶

type FrameMessage struct {
	BaseDataMessage
	// Frame is the frame being carried.
	Frame frames.Frame
	// Direction is the direction it should travel in on arrival.
	Direction processor.Direction
	// Bridge names the bridge it came through, or "" when it did not come
	// through one.
	Bridge string
}

FrameMessage carries a pipeline frame between workers, which is how a worker bridged into another's pipeline exchanges work with it.

type JSONSerializer ¶

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

JSONSerializer writes bus messages as JSON, tagging every value whose type cannot be read back from the JSON alone.

A value is written as it stands when JSON already describes it: a string, a number, a bool. Anything else is written as {"__type__":…,"__data__":…}, and the name is what the far end looks up to know what to build. Raw bytes are tagged too, base64 inside, since JSON has no way to carry them.

A type whose state the serializer cannot reach, because it is private or because its wire form is deliberately not its field layout, is handled by a TypeAdapter registered against it. The adapters for a conversation and its toolset are registered by default.

func NewJSONSerializer(types *TypeRegistry) *JSONSerializer

func (s *JSONSerializer) Deserialize(data []byte) (Message, error)

func (s *JSONSerializer) RegisterAdapter(sample any, a TypeAdapter)

func (s *JSONSerializer) Serialize(m Message) ([]byte, error)

type JobCancelMessage ¶

type JobCancelMessage struct {
	BaseSystemMessage
	// JobID identifies the job.
	JobID string
	// Reason describes why, and may be empty.
	Reason string
}

JobCancelMessage calls a job off. It is a system message, so it reaches the worker ahead of the work it is calling off.

type JobProgress ¶

type JobProgress struct {
	// JobID identifies the job.
	JobID string
	// Update is the progress being reported, and may be nil.
	Update map[string]any
}

JobProgress is the body of a message reporting progress on a running job.

func (p *JobProgress) Progress() *JobProgress

type JobRequestMessage ¶

type JobRequestMessage struct {
	BaseDataMessage
	// JobID identifies the job, and is what every later message about it
	// carries.
	JobID string
	// JobName names the kind of work, and may be empty.
	JobName string
	// Payload is the job's input, and may be nil.
	Payload map[string]any
}

JobRequestMessage asks a worker to carry out a job.

type JobResponse ¶

type JobResponse interface {
	Message
	// Result is what the message reports.
	Result() *JobResult
}

JobResponse reports how a job ended, whichever band it travels in.

type JobResponseMessage ¶

type JobResponseMessage struct {
	BaseDataMessage
	JobResult
}

JobResponseMessage reports how a job ended.

type JobResponseUrgentMessage ¶

type JobResponseUrgentMessage struct {
	BaseSystemMessage
	JobResult
}

JobResponseUrgentMessage reports how a job ended, ahead of the queued work.

type JobResult ¶

type JobResult struct {
	// JobID identifies the job.
	JobID string
	// Status is how it ended.
	Status jobcontext.JobStatus
	// Response is the job's output, and may be nil.
	Response map[string]any
}

JobResult is the body of a message reporting how a job ended.

func (r *JobResult) Result() *JobResult

type JobStreamDataMessage ¶

type JobStreamDataMessage struct {
	BaseDataMessage
	// JobID identifies the job.
	JobID string
	// Data is the item, and may be nil.
	Data map[string]any
}

JobStreamDataMessage carries one item of a job's result stream.

type JobStreamEndMessage ¶

type JobStreamEndMessage struct {
	BaseDataMessage
	// JobID identifies the job.
	JobID string
	// Data is what closes the stream, and may be nil.
	Data map[string]any
}

JobStreamEndMessage closes a job's result stream.

type JobStreamStartMessage ¶

type JobStreamStartMessage struct {
	BaseDataMessage
	// JobID identifies the job.
	JobID string
	// Data is what opens the stream, and may be nil.
	Data map[string]any
}

JobStreamStartMessage opens a stream of results for a job.

type JobUpdate ¶

type JobUpdate interface {
	Message
	// Progress is what the message reports.
	Progress() *JobProgress
}

JobUpdate reports progress on a running job, whichever band it travels in.

type JobUpdateMessage ¶

type JobUpdateMessage struct {
	BaseDataMessage
	JobProgress
}

JobUpdateMessage reports progress on a job still running.

type JobUpdateRequestMessage ¶

type JobUpdateRequestMessage struct {
	BaseDataMessage
	// JobID identifies the job.
	JobID string
}

JobUpdateRequestMessage asks for the current progress of a job.

type JobUpdateUrgentMessage ¶

type JobUpdateUrgentMessage struct {
	BaseSystemMessage
	JobProgress
}

JobUpdateUrgentMessage reports progress ahead of the queued work.

type LocalMessage ¶

type LocalMessage interface {
	Message
	// contains filtered or unexported methods
}

LocalMessage is a message that stays on this bus and is never forwarded to a remote one.

type Message ¶

type Message interface {
	// Source is the name of the worker or component that sent it.
	Source() string
	// Target is the name of the worker it is for, or "" to broadcast.
	Target() string
	// contains filtered or unexported methods
}

Message is one message carried by the bus.

type MessageFilter ¶

type MessageFilter interface {
	// AcceptsBusMessage reports whether this subscriber should be handed this
	// message. Returning false drops it for this subscriber alone; the others
	// still receive it.
	AcceptsBusMessage(m Message) bool
}

MessageFilter is implemented by a Subscriber that does not take every message. The bus asks before every delivery, and a subscriber that does not implement it takes everything.

type MessageSerializer ¶

type MessageSerializer interface {
	// Serialize converts a message to the bytes to send.
	Serialize(m Message) ([]byte, error)
	// Deserialize reconstructs the message those bytes carried.
	Deserialize(data []byte) (Message, error)
}

MessageSerializer converts bus messages to bytes and back. A bus that leaves the process holds one and uses it at each edge.

type Publisher ¶

type Publisher interface {
	Publish(ctx context.Context, m Message)
}

Publisher is what a concrete bus implements to carry a message on its transport. Bus.Send calls it for everything that is not local-only.

type SerializeFunc ¶

type SerializeFunc func(v any) (any, error)

SerializeFunc converts one value to its wire form. A TypeAdapter is handed one so it can serialize what its own fields hold without knowing how.

type Subscriber ¶

type Subscriber interface {
	// Name identifies this subscriber on the bus. Subscribing twice under the
	// same name is a no-op.
	Name() string
	// OnBusMessage handles one message. A panic in it is contained: the
	// subscriber keeps receiving.
	OnBusMessage(ctx context.Context, m Message)
}

Subscriber receives messages from a bus.

type SystemMessage ¶

type SystemMessage interface {
	Message
	// contains filtered or unexported methods
}

SystemMessage is a message delivered ahead of the queued data messages.

type TTSSpeakMessage ¶

type TTSSpeakMessage struct {
	BaseDataMessage
	// Text is what to say.
	Text string
	// AppendToContext reports whether what is said joins the conversation.
	AppendToContext bool
}

TTSSpeakMessage asks a worker to say something.

type TypeAdapter ¶

type TypeAdapter interface {
	// Serialize converts obj to a map, using serialize for whatever its own
	// fields hold that it cannot convert itself.
	Serialize(obj any, serialize SerializeFunc) (map[string]any, error)
	// Deserialize rebuilds the value from a map serialize produced.
	Deserialize(data map[string]any, deserialize DeserializeFunc) (any, error)
}

TypeAdapter converts values of one type to and from a plain map, for a type the serializer cannot take apart itself: one whose state is private, or whose wire form is deliberately not its field layout.

Register one with JSONSerializer.RegisterAdapter. Unlike upstream, the adapter is not told which type to build: it is registered against exactly one type and is the only thing that builds it.

type TypeRegistry ¶

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

TypeRegistry maps the type names on the wire to the values they name.

It is what stands in for reading a type out of a module path at run time, which is how the same machinery works in a dynamic language and is not available here. A type that may cross a bus registers itself, once, and the registry answers in both directions: the name to write for a value, and a fresh value to fill for a name.

A name is the type's own, without its package: "FrameMessage", "TranscriptionFrame". Two registered types may not share one, since the name is all the far end has to go on.

func NewTypeRegistry() *TypeRegistry

func (r *TypeRegistry) NameOf(v any) (string, bool)

func (r *TypeRegistry) Names() []string

func (r *TypeRegistry) New(name string) (any, bool)

func (r *TypeRegistry) Register(name string, sample any)

type WorkerErrorMessage ¶

type WorkerErrorMessage struct {
	BaseSystemMessage
	// Error describes the failure.
	Error string
}

WorkerErrorMessage reports a worker that failed, to everyone.

type WorkerLocalErrorMessage ¶

type WorkerLocalErrorMessage struct {
	BaseSystemMessage

	// Error describes the failure.
	Error string
	// contains filtered or unexported fields
}

WorkerLocalErrorMessage reports a worker that failed, to this process only.

type WorkerReadyMessage ¶

type WorkerReadyMessage struct {
	BaseDataMessage
	// Runner is the runner managing the worker.
	Runner string
	// Parent is the worker's parent, or "" for a root worker.
	Parent string
	// Active reports whether it is currently active.
	Active bool
	// Bridged reports whether it is bridged.
	Bridged bool
	// StartedAt is when it became ready, as a Unix timestamp, and zero when
	// unset.
	StartedAt float64
}

WorkerReadyMessage announces that a worker has started and can be addressed.

type WorkerRegistryMessage ¶

type WorkerRegistryMessage struct {
	BaseSystemMessage
	// Runner is the runner the snapshot belongs to.
	Runner string
	// Workers is the snapshot.
	Workers []registry.WorkerRegistryEntry
}

WorkerRegistryMessage announces which workers a runner knows about, so the runners on a bus can learn of each other's workers.