Documentation
¶
Overview ¶
Package pipeline connects frame processors into a chain and drives them.
A Pipeline links a sequence of processors between a source and a sink. A Task runs a pipeline for one session: it sends the StartFrame, pushes frames, and shuts the pipeline down on an EndFrame or CancelFrame. A Runner runs a Task and ends it on an interrupt signal.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Observer ¶ added in v0.0.4
Observer watches frames flowing through a pipeline without modifying them, to derive turn, latency or startup metrics, or to log the stream. The task reports each frame that reaches the pipeline's downstream end (with dir Downstream) or upstream end (with dir Upstream) to every registered observer.
Observers must be safe for concurrent use: the two ends run on separate goroutines, so OnFrame may be called from either.
type ParallelPipeline ¶
ParallelPipeline runs several sub-pipelines, the branches, concurrently. Every frame entering the parallel pipeline is fanned out to all branches, and frames emerging from the branches are merged back out. A frame that more than one branch forwards unchanged escapes only once: the merge deduplicates by frame id.
Lifecycle frames — StartFrame, EndFrame and CancelFrame — are synchronized. The parallel pipeline waits for every branch to process the frame before letting a single copy continue, and buffers any other frames a branch emits in the meantime (flushing them after a StartFrame, before an EndFrame or CancelFrame). This stops a fast branch from leaking an EndFrame and shutting downstream processors down while a slower branch still has output to flush, or from emitting data before every branch has been started.
A ParallelPipeline is itself a processor, so it nests inside a Pipeline.
The deduplication set retains the id of every frame that escapes the parallel pipeline for the lifetime of the session; this mirrors the upstream design and is bounded only by the session's frame count.
func NewParallel ¶
func NewParallel(branches ...[]processor.Processor) (*ParallelPipeline, error)
NewParallel builds a ParallelPipeline from one or more branches, each a list of processors connected in order. It returns an error if no branch is given.
func (*ParallelPipeline) Cleanup ¶
func (p *ParallelPipeline) Cleanup(ctx context.Context) error
Cleanup cleans up the parallel pipeline and every branch.
func (*ParallelPipeline) ProcessFrame ¶
func (p *ParallelPipeline) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error
ProcessFrame fans a frame out to every branch. For a lifecycle frame it first arms the synchronization counter, then blocks until every branch has processed the frame (or the context is canceled), so the next frame is not fanned out until the lifecycle frame has fully propagated.
type Pipeline ¶
Pipeline is a linear chain of processors. It wraps the chain with a source and a sink so frames can be fed in and observed at the edges, and is itself a processor, so pipelines can nest.
func New ¶
New builds a Pipeline from procs, connected in order. Frames pushed out of the chain's source and sink are forwarded to the pipeline's own neighbors.
func (*Pipeline) ProcessFrame ¶
ProcessFrame routes a frame into the chain: downstream frames enter at the source, upstream frames enter at the sink.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner runs a single task to completion and shuts it down gracefully on an interrupt signal (SIGINT or SIGTERM).
func NewRunner ¶
func NewRunner(opts ...RunnerOption) *Runner
NewRunner returns a Runner. By default it handles interrupt signals.
func (*Runner) Run ¶
Run runs task until it finishes. When signal handling is enabled, the first interrupt signal cancels the task (so the pipeline drains the CancelFrame and shuts down); the task keeps running on the parent context so cleanup can complete. If the parent context is canceled the task stops the same way.
type RunnerOption ¶
type RunnerOption func(*Runner)
RunnerOption configures a Runner.
func WithoutSignalHandling ¶
func WithoutSignalHandling() RunnerOption
WithoutSignalHandling disables the SIGINT/SIGTERM handling, leaving shutdown entirely to the caller's context.
type ServiceSwitcher ¶
type ServiceSwitcher struct {
*Pipeline
// contains filtered or unexported fields
}
ServiceSwitcher routes the pipeline through one of several interchangeable services at a time. Every service is started and kept warm, but only the active one receives data; the rest are gated off. Switching is manual (via SwitchTo or a SwitchServiceFrame) and, under SwitchFailover, automatic when the active service reports a non-fatal error.
It is built on a ParallelPipeline: each service becomes a branch wrapped in a pair of filters that pass lifecycle and system frames (so every service stays ready) but gate data frames on whether the service is active. A control processor in front consumes switch requests and watches for the errors that drive failover.
func NewServiceSwitcher ¶
func NewServiceSwitcher(services []processor.Processor, strategy SwitcherStrategy) (*ServiceSwitcher, error)
NewServiceSwitcher builds a switcher over services, the first of which starts active, using the given switching strategy.
func (*ServiceSwitcher) ActiveService ¶
func (s *ServiceSwitcher) ActiveService() processor.Processor
ActiveService returns the currently active service.
func (*ServiceSwitcher) OnSwitch ¶
func (s *ServiceSwitcher) OnSwitch(fn func(processor.Processor))
OnSwitch registers fn to be called whenever the active service changes.
type SwitchServiceFrame ¶
type SwitchServiceFrame struct {
frames.BaseControlFrame
// Service is the service to activate; it must belong to the switcher.
Service processor.Processor
}
SwitchServiceFrame requests that a ServiceSwitcher make Service active. Queue it downstream into the pipeline, or call ServiceSwitcher.SwitchTo directly.
func NewSwitchServiceFrame ¶
func NewSwitchServiceFrame(svc processor.Processor) *SwitchServiceFrame
NewSwitchServiceFrame builds a SwitchServiceFrame targeting svc.
type SwitcherStrategy ¶
type SwitcherStrategy int
SwitcherStrategy selects how a ServiceSwitcher changes its active service.
const ( // SwitchManual changes the active service only on an explicit request. SwitchManual SwitcherStrategy = iota // SwitchFailover additionally moves to the next service when the active one // reports a non-fatal error. SwitchFailover )
type Task ¶
type Task struct {
// contains filtered or unexported fields
}
Task runs a pipeline for a single session. It drives the lifecycle: it sends the StartFrame, waits for the pipeline to be ready, pushes queued frames, and shuts the pipeline down once an EndFrame or CancelFrame has traveled all the way through.
func NewTask ¶
func NewTask(pipe processor.Processor, params TaskParams) *Task
NewTask wraps pipe in a Task. pipe is usually a *Pipeline but may be any processor.
func (*Task) Cancel ¶
func (t *Task) Cancel()
Cancel stops the pipeline immediately by queueing a CancelFrame.
func (*Task) HasFinished ¶
HasFinished reports whether the task has finished running.
func (*Task) QueueFrame ¶
QueueFrame queues a frame to be pushed downstream through the pipeline.
func (*Task) QueueFrames ¶
QueueFrames queues several frames to be pushed downstream, in order.
func (*Task) Run ¶
Run sets up the pipeline and drives it until an EndFrame or CancelFrame completes its journey through the pipeline, or ctx is canceled. It then cleans up the pipeline. Run blocks until the pipeline has finished.
func (*Task) StopWhenDone ¶
func (t *Task) StopWhenDone()
StopWhenDone schedules the pipeline to stop once all queued frames have been processed, by queueing an EndFrame.
type TaskParams ¶
type TaskParams struct {
// Clock is the pipeline clock; a system clock is used when nil.
Clock clock.Clock
// AudioInSampleRate is the StartFrame input sample rate; default 16000.
AudioInSampleRate int
// AudioOutSampleRate is the StartFrame output sample rate; default 24000.
AudioOutSampleRate int
// EnableMetrics enables performance-metrics collection across the pipeline.
EnableMetrics bool
// EnableUsageMetrics enables usage-metrics collection (e.g. LLM token usage)
// across the pipeline.
EnableUsageMetrics bool
// OnReachedDownstream, if set, is called for every frame that reaches the
// end of the pipeline.
OnReachedDownstream func(frames.Frame)
// OnReachedUpstream, if set, is called for every frame that reaches the
// start of the pipeline.
OnReachedUpstream func(frames.Frame)
// Observers watch every frame reaching either end of the pipeline. They are
// notified after the OnReached callbacks.
Observers []Observer
}
TaskParams configures a Task.
Source Files