mapping

Documentation

Overview

Package mapping provides a parser for the full bloblang mapping spec.

Index

• func LineAndColOf(input, clip []rune) (line, col int)
• type Assignment
• type AssignmentContext
• type Executor
  • func NewExecutor(annotation string, input []rune, maps map[string]query.Function, ...) *Executor
  • func (e *Executor) Annotation() string
  • func (e *Executor) AssignmentTargets() []TargetPath
  • func (e *Executor) Exec(ctx query.FunctionContext) (any, error)
  • func (e *Executor) ExecOnto(ctx query.FunctionContext, onto AssignmentContext) error
  • func (e *Executor) MapOnto(part *message.Part, index int, msg Message) (*message.Part, error)
  • func (e *Executor) MapPart(index int, msg Message) (*message.Part, error)
  • func (e *Executor) Maps() map[string]query.Function
  • func (e *Executor) QueryPart(index int, msg Message) (bool, error)
  • func (e *Executor) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)
  • func (e *Executor) SetMaxMapRecursion(m int)
  • func (e *Executor) ToBytes(ctx query.FunctionContext) ([]byte, error)
  • func (e *Executor) ToString(ctx query.FunctionContext) (string, error)
• type JSONAssignment
  • func NewJSONAssignment(path ...string) *JSONAssignment
  • func (j *JSONAssignment) Apply(val any, ctx AssignmentContext) error
  • func (j *JSONAssignment) Target() TargetPath
• type Message
• type MetaAssignment
  • func NewMetaAssignment(key *string) *MetaAssignment
  • func (m *MetaAssignment) Apply(val any, ctx AssignmentContext) error
  • func (m *MetaAssignment) Target() TargetPath
• type RootLevelIfStatement
  • func NewRootLevelIfStatement(input []rune) *RootLevelIfStatement
  • func (r *RootLevelIfStatement) Add(query query.Function, statements ...Statement) *RootLevelIfStatement
  • func (r *RootLevelIfStatement) AssignmentTargets() []TargetPath
  • func (r *RootLevelIfStatement) Execute(fnContext query.FunctionContext, asContext AssignmentContext) error
  • func (r *RootLevelIfStatement) Input() []rune
  • func (r *RootLevelIfStatement) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)
• type SingleStatement
  • func NewSingleStatement(input []rune, assignment Assignment, query query.Function) *SingleStatement
  • func (s *SingleStatement) AssignmentTargets() []TargetPath
  • func (s *SingleStatement) Execute(fnContext query.FunctionContext, asContext AssignmentContext) error
  • func (s *SingleStatement) Input() []rune
  • func (s *SingleStatement) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)
• type Statement
• type TargetPath
  • func NewTargetPath(t TargetType, path ...string) TargetPath
• type TargetType
• type VarAssignment
  • func NewVarAssignment(name string) *VarAssignment
  • func (v *VarAssignment) Apply(val any, ctx AssignmentContext) error
  • func (v *VarAssignment) Target() TargetPath

Functions

func LineAndColOf

func LineAndColOf(input, clip []rune) (line, col int)

LineAndColOf returns the line and column position of a tailing clip from an input.

Types

type Assignment

type Assignment interface {
	Apply(value any, ctx AssignmentContext) error
	Target() TargetPath
}

Assignment represents a way of assigning a queried value to something within an assignment context. This could be a Benthos message, a variable, a metadata field, etc.

type AssignmentContext

type AssignmentContext struct {
	Vars  map[string]any
	Meta  metaMsg
	Value *any
}

AssignmentContext contains references to all potential assignment destinations of a given mapping.

type Executor

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

Executor is a parsed bloblang mapping that can be executed on a Benthos message.

func NewExecutor

func NewExecutor(annotation string, input []rune, maps map[string]query.Function, statements ...Statement) *Executor

NewExecutor initialises a new mapping executor from a map of query functions, and a list of assignments to be executed on each mapping. The input parameter is an optional slice pointing to the parsed expression that created the executor.

func (*Executor) Annotation

func (e *Executor) Annotation() string

Annotation returns a string annotation that describes the mapping executor.

func (*Executor) AssignmentTargets

func (e *Executor) AssignmentTargets() []TargetPath

AssignmentTargets returns a slice of all targets assigned to by statements within the mapping.

func (*Executor) Exec

func (e *Executor) Exec(ctx query.FunctionContext) (any, error)

Exec this function with a context struct.

func (*Executor) ExecOnto

func (e *Executor) ExecOnto(ctx query.FunctionContext, onto AssignmentContext) error

ExecOnto a provided assignment context.

func (*Executor) MapOnto

func (e *Executor) MapOnto(part *message.Part, index int, msg Message) (*message.Part, error)

MapOnto maps into an existing message part, where mappings are appended to the message rather than being used to construct a new message.

func (*Executor) MapPart

func (e *Executor) MapPart(index int, msg Message) (*message.Part, error)

MapPart executes the bloblang mapping on a particular message index of a batch. The message is parsed as a JSON document in order to provide the mapping context. Returns an error if any stage of the mapping fails to execute.

A resulting mapped message part is returned, unless the mapping results in a value.Delete value, in which case nil is returned and the part should be discarded.

func (*Executor) Maps

func (e *Executor) Maps() map[string]query.Function

Maps returns any map definitions contained within the mapping.

func (*Executor) QueryPart

func (e *Executor) QueryPart(index int, msg Message) (bool, error)

QueryPart executes the bloblang mapping on a particular message index of a batch. The message is parsed as a JSON document in order to provide the mapping context. The result of the mapping is expected to be a boolean value at the root, this is not the case, or if any stage of the mapping fails to execute, an error is returned.

func (*Executor) QueryTargets

func (e *Executor) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)

QueryTargets returns a slice of all targets referenced by queries within the mapping.

func (*Executor) SetMaxMapRecursion

func (e *Executor) SetMaxMapRecursion(m int)

SetMaxMapRecursion configures the maximum recursion allowed for maps, if the execution of this mapping matches this number of recursive map calls the mapping will error out.

func (*Executor) ToBytes

func (e *Executor) ToBytes(ctx query.FunctionContext) ([]byte, error)

ToBytes executes this function for a message of a batch and returns the result marshalled into a byte slice.

func (*Executor) ToString

func (e *Executor) ToString(ctx query.FunctionContext) (string, error)

ToString executes this function for a message of a batch and returns the result marshalled into a string.

type JSONAssignment

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

JSONAssignment creates a path within the structured message and assigns it a value.

func NewJSONAssignment

func NewJSONAssignment(path ...string) *JSONAssignment

NewJSONAssignment creates a new JSON assignment.

func (*JSONAssignment) Apply

func (j *JSONAssignment) Apply(val any, ctx AssignmentContext) error

Apply a value to the target JSON path.

func (*JSONAssignment) Target

func (j *JSONAssignment) Target() TargetPath

Target returns a representation of what the assignment targets.

type Message

type Message interface {
	Get(p int) *message.Part
	Len() int
}

Message is an interface type to be given to a query function, it allows the function to resolve fields and metadata from a message.

type MetaAssignment

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

MetaAssignment assigns a value to a metadata key of a message. If the key is omitted and the value is an object then the metadata of the message is reset to the contents of the value.

func NewMetaAssignment

func NewMetaAssignment(key *string) *MetaAssignment

NewMetaAssignment creates a new meta assignment.

func (*MetaAssignment) Apply

func (m *MetaAssignment) Apply(val any, ctx AssignmentContext) error

Apply a value to a metadata key.

func (*MetaAssignment) Target

func (m *MetaAssignment) Target() TargetPath

Target returns a representation of what the assignment targets.

type RootLevelIfStatement

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

RootLevelIfStatement describes an isolated conditional mapping statement.

func NewRootLevelIfStatement

func NewRootLevelIfStatement(input []rune) *RootLevelIfStatement

NewRootLevelIfStatement initialises a new conditional mapping statement. The input parameter is a slice pointing to the parsed expression that created the statement.

func (*RootLevelIfStatement) Add

func (r *RootLevelIfStatement) Add(query query.Function, statements ...Statement) *RootLevelIfStatement

Add adds query statement pairs to the root level if statement.

func (*RootLevelIfStatement) AssignmentTargets

func (r *RootLevelIfStatement) AssignmentTargets() []TargetPath

AssignmentTargets returns a representation of what the underlying conditional mapping statement targets.

func (*RootLevelIfStatement) Execute

func (r *RootLevelIfStatement) Execute(fnContext query.FunctionContext, asContext AssignmentContext) error

Execute executes this statement if the underlying condition evaluates to true.

func (*RootLevelIfStatement) Input

func (r *RootLevelIfStatement) Input() []rune

Input returns the underlying parsed expression of this conditional mapping statement.

func (*RootLevelIfStatement) QueryTargets

func (r *RootLevelIfStatement) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)

QueryTargets returns the query targets for the underlying conditional mapping statement.

type SingleStatement

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

SingleStatement describes an isolated mapping statement, where the result of a query function is to be mapped according to an Assignment.

func NewSingleStatement

func NewSingleStatement(input []rune, assignment Assignment, query query.Function) *SingleStatement

NewSingleStatement initialises a new mapping statement from an Assignment and query.Function. The input parameter is an optional slice pointing to the parsed expression that created the statement.

func (*SingleStatement) AssignmentTargets

func (s *SingleStatement) AssignmentTargets() []TargetPath

AssignmentTargets returns a representation of what the underlying assignment targets.

func (*SingleStatement) Execute

func (s *SingleStatement) Execute(fnContext query.FunctionContext, asContext AssignmentContext) error

Execute executes this statement and applies the result onto the assigned destination.

func (*SingleStatement) Input

func (s *SingleStatement) Input() []rune

Input returns the underlying parsed expression of this statement.

func (*SingleStatement) QueryTargets

func (s *SingleStatement) QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)

QueryTargets returns the query targets for the underlying query.

type Statement

type Statement interface {
	QueryTargets(ctx query.TargetsContext) (query.TargetsContext, []query.TargetPath)
	AssignmentTargets() []TargetPath
	Input() []rune
	Execute(fnContext query.FunctionContext, asContext AssignmentContext) error
}

Statement represents a bloblang mapping statement.

type TargetPath

type TargetPath struct {
	Type TargetType
	Path []string
}

TargetPath represents a target type and segmented path that a query function references. An empty path indicates the root of the type is targeted.

func NewTargetPath

func NewTargetPath(t TargetType, path ...string) TargetPath

NewTargetPath constructs a new target path from a type and zero or more path segments.

type TargetType

type TargetType int

TargetType represents a mapping target type, which is a destination for a query result to be mapped into a message.

const (
	TargetMetadata TargetType = iota
	TargetValue
	TargetVariable
)

TargetTypes.

type VarAssignment

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

VarAssignment creates a variable and assigns it a value.

func NewVarAssignment

func NewVarAssignment(name string) *VarAssignment

NewVarAssignment creates a new variable assignment.

func (*VarAssignment) Apply

func (v *VarAssignment) Apply(val any, ctx AssignmentContext) error

Apply a value to a variable.

func (*VarAssignment) Target

func (v *VarAssignment) Target() TargetPath

Target returns a representation of what the assignment targets.