processor

Documentation

Overview

Package processor defines the frame processor: the building block of a jargo pipeline. Processors link into a chain, receive frames, process them, and push them on to the next or previous processor. Each processor handles system frames with priority, processes data and control frames in order on its own goroutine, and can be interrupted.

Index

• type Base
  • func New(name string, self Processor, opts ...Option) *Base
  • func (b *Base) Cleanup(ctx context.Context) error
  • func (b *Base) Clock() clock.Clock
  • func (b *Base) ID() uint64
  • func (b *Base) Link(next Processor)
  • func (b *Base) MetricsEnabled() bool
  • func (b *Base) Name() string
  • func (b *Base) Next() Processor
  • func (b *Base) Prev() Processor
  • func (b *Base) ProcessFrame(ctx context.Context, f frames.Frame, dir Direction) error
  • func (b *Base) PushError(ctx context.Context, msg string, err error, fatal bool)
  • func (b *Base) PushFrame(ctx context.Context, f frames.Frame, dir Direction) error
  • func (b *Base) QueueFrame(ctx context.Context, f frames.Frame, dir Direction) error
  • func (b *Base) Setup(ctx context.Context, s Setup) error
  • func (b *Base) UsageMetricsEnabled() bool
• type Direction
  • func (d Direction) String() string
• type FilterFunc
• type FunctionFilter
  • func NewFunctionFilter(name string, dir Direction, allow FilterFunc) *FunctionFilter
  • func (f *FunctionFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error
• type HandlerFunc
• type Option
  • func WithDirectMode() Option
• type Processor
  • func NewSink(name string, downstream HandlerFunc) Processor
  • func NewSource(name string, upstream HandlerFunc) Processor
• type Setup

Types

type Base

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

Base implements Processor. Embed it in a concrete processor and pass the concrete value as self so the base can dispatch to the overridden ProcessFrame:

type Echo struct{ *processor.Base }

func NewEcho() *Echo {
    e := &Echo{}
    e.Base = processor.New("Echo", e)
    return e
}

func (e *Echo) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error {
    if err := e.Base.ProcessFrame(ctx, f, dir); err != nil {
        return err
    }
    return e.PushFrame(ctx, f, dir)
}

func New

func New(name string, self Processor, opts ...Option) *Base

New builds a Base named name. self is the embedding processor, used to dispatch to its ProcessFrame; pass nil for a plain pass-through that does not override ProcessFrame.

func (*Base) Cleanup

func (b *Base) Cleanup(ctx context.Context) error

Cleanup implements Processor. It stops the process and input goroutines.

func (*Base) Clock

func (b *Base) Clock() clock.Clock

Clock returns the pipeline clock, available after Setup.

func (*Base) ID

func (b *Base) ID() uint64

ID implements Processor.

func (*Base) Link

func (b *Base) Link(next Processor)

Link implements Processor.

func (*Base) MetricsEnabled

func (b *Base) MetricsEnabled() bool

MetricsEnabled reports whether performance-metrics collection was enabled by the StartFrame. It is valid once the processor has received its StartFrame.

func (*Base) Name

func (b *Base) Name() string

Name implements Processor.

func (*Base) Next

func (b *Base) Next() Processor

Next implements Processor.

func (*Base) Prev

func (b *Base) Prev() Processor

Prev implements Processor.

func (*Base) ProcessFrame

func (b *Base) ProcessFrame(ctx context.Context, f frames.Frame, dir Direction) error

ProcessFrame implements Processor. It handles the system frames that drive a processor's lifecycle: StartFrame, InterruptionFrame and CancelFrame. A concrete processor overrides this, calls the base first, then forwards the frame with PushFrame.

func (*Base) PushError

func (b *Base) PushError(ctx context.Context, msg string, err error, fatal bool)

PushError builds an ErrorFrame for msg and pushes it upstream.

func (*Base) PushFrame

func (b *Base) PushFrame(ctx context.Context, f frames.Frame, dir Direction) error

PushFrame implements Processor. It forwards a frame to the neighbor in dir. Frames pushed before the processor has received its StartFrame are dropped.

func (*Base) QueueFrame

func (b *Base) QueueFrame(ctx context.Context, f frames.Frame, dir Direction) error

QueueFrame implements Processor.

func (*Base) Setup

func (b *Base) Setup(ctx context.Context, s Setup) error

Setup implements Processor. It stores shared components and starts the input goroutine (unless the processor is in direct mode).

func (*Base) UsageMetricsEnabled

func (b *Base) UsageMetricsEnabled() bool

UsageMetricsEnabled reports whether usage-metrics collection was enabled by the StartFrame. It is valid once the processor has received its StartFrame.

type Direction

type Direction int

Direction is the direction a frame flows through the pipeline.

const (
	// Downstream is the direction from input toward output.
	Downstream Direction = iota
	// Upstream is the direction from output back toward input.
	Upstream
)

func (Direction) String

func (d Direction) String() string

String returns "downstream" or "upstream".

type FilterFunc

type FilterFunc func(frames.Frame) bool

FilterFunc reports whether a frame is allowed to pass through a FunctionFilter.

type FunctionFilter

type FunctionFilter struct {
	*Base
	// contains filtered or unexported fields
}

FunctionFilter forwards frames, dropping those that travel in a chosen direction and for which a predicate returns false; frames in the other direction always pass. It runs in direct mode, deciding on the caller's goroutine, and is the building block a ServiceSwitcher uses to gate a branch on or off.

func NewFunctionFilter

func NewFunctionFilter(name string, dir Direction, allow FilterFunc) *FunctionFilter

NewFunctionFilter builds a filter that gates frames moving in dir using allow.

func (*FunctionFilter) ProcessFrame

func (f *FunctionFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

ProcessFrame drops a frame moving in the gated direction when the predicate rejects it, and forwards everything else.

type HandlerFunc

type HandlerFunc func(ctx context.Context, f frames.Frame, dir Direction) error

HandlerFunc handles a frame that reaches the edge of a pipeline. A source uses it for upstream frames, a sink for downstream frames.

type Option

type Option func(*Base)

Option configures a Base at construction.

func WithDirectMode

func WithDirectMode() Option

WithDirectMode makes a processor process frames immediately on the caller's goroutine instead of queueing them. It is used for routing processors (a pipeline and its source and sink) that only forward frames.

type Processor

type Processor interface {
	// ID is a process-unique identifier for this processor.
	ID() uint64
	// Name is a human-readable label, "<name>#<id>".
	Name() string

	// Next is the downstream processor, or nil.
	Next() Processor
	// Prev is the upstream processor, or nil.
	Prev() Processor
	// Link sets next as this processor's downstream neighbor and this
	// processor as next's upstream neighbor.
	Link(next Processor)

	// Setup wires the processor with shared components and starts its
	// goroutines. It must be called before frames are queued.
	Setup(ctx context.Context, s Setup) error
	// Cleanup stops the processor's goroutines and releases resources.
	Cleanup(ctx context.Context) error

	// QueueFrame hands a frame to this processor for processing.
	QueueFrame(ctx context.Context, f frames.Frame, dir Direction) error
	// ProcessFrame processes a frame. The base implementation handles system
	// lifecycle frames; concrete processors override it and call the base
	// first.
	ProcessFrame(ctx context.Context, f frames.Frame, dir Direction) error
	// PushFrame sends a frame to the neighboring processor in dir.
	PushFrame(ctx context.Context, f frames.Frame, dir Direction) error
}

Processor is a node in a pipeline. Concrete processors embed *Base, which provides every method here except a custom ProcessFrame.

func NewSink

func NewSink(name string, downstream HandlerFunc) Processor

NewSink returns a pipeline sink. Downstream frames reaching it are passed to downstream; upstream frames are forwarded back along the chain.

func NewSource

func NewSource(name string, upstream HandlerFunc) Processor

NewSource returns a pipeline source. Upstream frames reaching it are passed to upstream; downstream frames are forwarded along the chain.

type Setup

type Setup struct {
	// Clock is the pipeline clock used for timing.
	Clock clock.Clock
}

Setup carries the shared components a processor needs, propagated down the pipeline when it is set up.