core

Documentation

Index

• func Default() spec.ExpressionFactory
• func NewExpressionFactory() spec.ExpressionFactory
• func NoopCallback(ctx context.Context, err error) error
• type Batch
• type Component
• type ComponentContext
• type Expression
• type ExpressionContext
  • func MessageExpressionContext(msg spec.Message) ExpressionContext
• type ExpressionFactory
• type ExpressionSegment
• type FieldEvaluator
  • func NewFieldEvaluator(field string) (*FieldEvaluator, error)
  • func (fe *FieldEvaluator) Eval(batch spec.Batch, index int) (interface{}, error)
  • func (fe *FieldEvaluator) EvalBool(batch spec.Batch, index int) (bool, error)
  • func (fe *FieldEvaluator) EvalFloat64(batch spec.Batch, index int) (float64, error)
  • func (fe *FieldEvaluator) EvalInt(batch spec.Batch, index int) (int, error)
  • func (fe *FieldEvaluator) EvalString(batch spec.Batch, index int) (string, error)
  • func (fe *FieldEvaluator) GetExpressionCount() int
  • func (fe *FieldEvaluator) GetOriginal() string
  • func (fe *FieldEvaluator) HasExpressions() bool
• type Input
• type Message
• type MessageFactory
• type Output
• type ProcessedCallback
• type Registry
• type System

Functions

func Default

func Default() spec.ExpressionFactory

Default creates a default expression factory using expr-lang

func NewExpressionFactory

func NewExpressionFactory() spec.ExpressionFactory

NewExpressionFactory creates a new expression factory using expr-lang

func NoopCallback

func NoopCallback(ctx context.Context, err error) error

Types

type Batch

type Batch interface {
	Messages() iter.Seq2[int, Message]
	Append(msg Message)
}

type Component

type Component interface {
	Init(ctx ComponentContext) error
	Close(ctx ComponentContext) error
}

type ComponentContext

type ComponentContext interface {
	ExpressionFactory
	MessageFactory

	Context() context.Context

	Input(name string) (Input, error)
	Output(name string) (Output, error)
}

type Expression

type Expression = spec.Expression

type ExpressionContext

type ExpressionContext = spec.ExpressionContext

func MessageExpressionContext

func MessageExpressionContext(msg spec.Message) ExpressionContext

MessageExpressionContext creates an expression context from a message. It exposes the message's raw content and metadata for use in expressions. Deprecated: Use spec.MessageExpressionContext instead

type ExpressionFactory

type ExpressionFactory = spec.ExpressionFactory

Legacy interfaces - deprecated in favor of spec package equivalents

type ExpressionSegment

type ExpressionSegment struct {
	IsExpression bool
	Text         string      // For literal text
	Program      *vm.Program // For compiled expressions
	RawExpr      string      // Original expression text (for error reporting)
}

ExpressionSegment represents a part of the field - either literal text or a compiled expression

type FieldEvaluator

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

FieldEvaluator holds precompiled expressions and can evaluate them against multiple data sets

func NewFieldEvaluator

func NewFieldEvaluator(field string) (*FieldEvaluator, error)

NewFieldEvaluator creates a new FieldEvaluator by parsing and precompiling expressions in the field

func (*FieldEvaluator) Eval

func (fe *FieldEvaluator) Eval(batch spec.Batch, index int) (interface{}, error)

Eval evaluates the precompiled field against the provided batch and index, returns the result

func (*FieldEvaluator) EvalBool

func (fe *FieldEvaluator) EvalBool(batch spec.Batch, index int) (bool, error)

EvalBool evaluates the field and returns the result as a boolean. Supports conversion from booleans, numeric types (0 = false, non-zero = true), and strings ("true"/"1"/"yes"/"on" = true, "false"/"0"/"no"/"off"/"" = false). Returns error if the result cannot be converted to bool.

func (*FieldEvaluator) EvalFloat64

func (fe *FieldEvaluator) EvalFloat64(batch spec.Batch, index int) (float64, error)

EvalFloat64 evaluates the field and returns the result as a float64. Supports conversion from numeric types, strings (if parseable), and booleans. Returns error if the result cannot be converted to float64.

func (*FieldEvaluator) EvalInt

func (fe *FieldEvaluator) EvalInt(batch spec.Batch, index int) (int, error)

EvalInt evaluates the field and returns the result as an integer. Supports conversion from numeric types, strings (if parseable), and booleans. Returns error if the result cannot be converted to int.

func (*FieldEvaluator) EvalString

func (fe *FieldEvaluator) EvalString(batch spec.Batch, index int) (string, error)

EvalString evaluates the field and returns the result as a string. All types are converted to their string representation using fmt.Sprintf.

func (*FieldEvaluator) GetExpressionCount

func (fe *FieldEvaluator) GetExpressionCount() int

GetExpressionCount returns the number of expressions in the field

func (*FieldEvaluator) GetOriginal

func (fe *FieldEvaluator) GetOriginal() string

GetOriginal returns the original field string

func (*FieldEvaluator) HasExpressions

func (fe *FieldEvaluator) HasExpressions() bool

HasExpressions returns true if the field contains any expressions

type Input

type Input interface {
	Component
	Read(ctx ComponentContext) (Batch, ProcessedCallback, error)
}

type Message

type Message interface {
	SetMetadata(key string, value any)
	SetRaw(b []byte)

	Raw() ([]byte, error)
	Metadata() iter.Seq2[string, any]
}

type MessageFactory

type MessageFactory interface {
	NewBatch() Batch
	NewMessage() Message
}

type Output

type Output interface {
	Component
	Write(ctx ComponentContext, batch Batch) error
}

type ProcessedCallback

type ProcessedCallback func(ctx context.Context, err error) error

ProcessedCallback is a function signature for a callback to be called after a message or message batch has been processed. The provided error indicates whether the processing was successful. TODO: add extra information on what happens if an error is returned

type Registry

type Registry interface {
	RegisterInput(name string, input Input) error
	GetInput(name string) (Input, error)

	RegisterOutput(name string, output Output) error
	GetOutput(name string) (Output, error)
}

type System

type System interface {
	Schema() string
	Connect(ctx context.Context) error
	Close(ctx context.Context) error
	Client() any
}