stateful

Documentation

Index

• Constants
• Variables
• func Exce(evalCtx ruleql.Context, expr ruleql.Expr, msg stream.PublishMessage) error
• func Execute(c ruleql.Context, expr string) []byte
• func NewStateManager() *stateManager
• type AggregateAcc
• type AggregateCallExpr
• type AggregateData
  • func NewAvgAggregateFunc() AggregateData
  • func NewCountAggregateFunc() AggregateData
  • func NewMaxAggregateFunc() AggregateData
  • func NewMinAggregateFunc() AggregateData
  • func NewSumAggregateFunc() AggregateData
• type AggregateFunc
• type AvgAggregateData
  • func (fn *AvgAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node
  • func (fn *AvgAggregateData) Clean() error
  • func (fn *AvgAggregateData) Result() ruleql.Node
  • func (fn *AvgAggregateData) Value(key string) ruleql.Node
• type Context
  • func NewMessageContext(message stream.PublishMessage) Context
• type ContextAggCallableFunc
• type CountAggregateData
  • func (fn *CountAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node
  • func (fn *CountAggregateData) Clean() error
  • func (fn *CountAggregateData) Result() ruleql.Node
  • func (fn *CountAggregateData) Value(key string) ruleql.Node
• type MaxAggregateData
  • func (fn *MaxAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node
  • func (fn *MaxAggregateData) Clean() error
  • func (fn *MaxAggregateData) Result() ruleql.Node
  • func (fn *MaxAggregateData) Value(key string) ruleql.Node
• type MinAggregateData
  • func (fn *MinAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node
  • func (fn *MinAggregateData) Clean() error
  • func (fn *MinAggregateData) Result() ruleql.Node
  • func (fn *MinAggregateData) Value(key string) ruleql.Node
• type Node
• type State
• type StreamOperator
  • func New(ctx context.Context, rule *v1.RuleQL, stateFunc api.StateFunc) (*StreamOperator, error)
  • func (this *StreamOperator) Exce(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error
  • func (this *StreamOperator) ExceStateful(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error
  • func (this *StreamOperator) ExceStateless(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error
  • func (this *StreamOperator) Filter(ctx context.Context, evalCtx ruleql.Context, message stream.PublishMessage) bool
  • func (this *StreamOperator) ID() string
  • func (this *StreamOperator) Invoke(ctx context.Context, evalCtx ruleql.Context, msg stream.PublishMessage) error
• type SumAggregateData
  • func (fn *SumAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node
  • func (fn *SumAggregateData) Clean() error
  • func (fn *SumAggregateData) Result() ruleql.Node
  • func (fn *SumAggregateData) Value(key string) ruleql.Node
• type WindowState
  • func (ws *WindowState) Exce(ctx functions.Context, operator *WindowsOperator) error
  • func (ws *WindowState) GetState() (state stream.Message)
  • func (ws *WindowState) Key() string
  • func (ws *WindowState) OnTrigger()
  • func (ws *WindowState) SetState(state stream.Message)
  • func (ws *WindowState) SetTriggerCallback(TriggerFunc api.StateFunc)
• type WindowsOperator
  • func NewWindowsOperator(expr ruleql.Expr, window *ruleql.WindowExpr, funcs map[string]*ruleql.CallExpr) *WindowsOperator

Constants

const ListKey = "x-state-list"

Variables

var (
	UNDEFINED_RESULT = ruleql.UNDEFINED_RESULT
)

Functions

func Exce

func Exce(evalCtx ruleql.Context, expr ruleql.Expr, msg stream.PublishMessage) error

func Execute

func Execute(c ruleql.Context, expr string) []byte

Value get value from context

func NewStateManager

func NewStateManager() *stateManager

Types

type AggregateAcc

type AggregateAcc func() AggregateData

type AggregateCallExpr

type AggregateCallExpr struct {
	*ruleql.CallExpr
	// contains filtered or unexported fields
}

type AggregateData

type AggregateData interface {
	ruleql.Context
	Result() ruleql.Node
	Clean() error
}

func NewAvgAggregateFunc

func NewAvgAggregateFunc() AggregateData

func NewCountAggregateFunc

func NewCountAggregateFunc() AggregateData

func NewMaxAggregateFunc

func NewMaxAggregateFunc() AggregateData

func NewMinAggregateFunc

func NewMinAggregateFunc() AggregateData

func NewSumAggregateFunc

func NewSumAggregateFunc() AggregateData

type AggregateFunc

type AggregateFunc func(evalCtx functions.Context, state *WindowState) error

type AvgAggregateData

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

func (*AvgAggregateData) Call

func (fn *AvgAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node

func (*AvgAggregateData) Clean

func (fn *AvgAggregateData) Clean() error

func (*AvgAggregateData) Result

func (fn *AvgAggregateData) Result() ruleql.Node

func (*AvgAggregateData) Value

func (fn *AvgAggregateData) Value(key string) ruleql.Node

type Context

type Context = ruleql.Context

func NewMessageContext

func NewMessageContext(message stream.PublishMessage) Context

NewJSONContext new context from json

type ContextAggCallableFunc

type ContextAggCallableFunc func(hash string, args ...Node) Node

type CountAggregateData

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

func (*CountAggregateData) Call

func (fn *CountAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node

func (*CountAggregateData) Clean

func (fn *CountAggregateData) Clean() error

func (*CountAggregateData) Result

func (fn *CountAggregateData) Result() ruleql.Node

func (*CountAggregateData) Value

func (fn *CountAggregateData) Value(key string) ruleql.Node

type MaxAggregateData

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

func (*MaxAggregateData) Call

func (fn *MaxAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node

func (*MaxAggregateData) Clean

func (fn *MaxAggregateData) Clean() error

func (*MaxAggregateData) Result

func (fn *MaxAggregateData) Result() ruleql.Node

func (*MaxAggregateData) Value

func (fn *MaxAggregateData) Value(key string) ruleql.Node

type MinAggregateData

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

func (*MinAggregateData) Call

func (fn *MinAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node

func (*MinAggregateData) Clean

func (fn *MinAggregateData) Clean() error

func (*MinAggregateData) Result

func (fn *MinAggregateData) Result() ruleql.Node

func (*MinAggregateData) Value

func (fn *MinAggregateData) Value(key string) ruleql.Node

type Node

type Node = ruleql.Node

type State

type State stream.Message

type StreamOperator

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

func New

func New(ctx context.Context, rule *v1.RuleQL, stateFunc api.StateFunc) (*StreamOperator, error)

New returns a new StreamOperator operator

func (*StreamOperator) Exce

func (this *StreamOperator) Exce(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error

Update message

func (*StreamOperator) ExceStateful

func (this *StreamOperator) ExceStateful(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error

func (*StreamOperator) ExceStateless

func (this *StreamOperator) ExceStateless(ctx context.Context, evalCtx ruleql.Context, msg stream.Message) error

func (*StreamOperator) Filter

func (this *StreamOperator) Filter(ctx context.Context, evalCtx ruleql.Context, message stream.PublishMessage) bool

Filter message

func (*StreamOperator) ID

func (this *StreamOperator) ID() string

func (*StreamOperator) Invoke

func (this *StreamOperator) Invoke(ctx context.Context, evalCtx ruleql.Context, msg stream.PublishMessage) error

type SumAggregateData

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

func (*SumAggregateData) Call

func (fn *SumAggregateData) Call(expr *ruleql.CallExpr, args []ruleql.Node) ruleql.Node

func (*SumAggregateData) Clean

func (fn *SumAggregateData) Clean() error

func (*SumAggregateData) Result

func (fn *SumAggregateData) Result() ruleql.Node

func (*SumAggregateData) Value

func (fn *SumAggregateData) Value(key string) ruleql.Node

type WindowState

type WindowState struct {
	TriggerFunc api.StateFunc
	// contains filtered or unexported fields
}

func (*WindowState) Exce

func (ws *WindowState) Exce(ctx functions.Context, operator *WindowsOperator) error

func (*WindowState) GetState

func (ws *WindowState) GetState() (state stream.Message)

func (*WindowState) Key

func (ws *WindowState) Key() string

func (*WindowState) OnTrigger

func (ws *WindowState) OnTrigger()

func (*WindowState) SetState

func (ws *WindowState) SetState(state stream.Message)

func (*WindowState) SetTriggerCallback

func (ws *WindowState) SetTriggerCallback(TriggerFunc api.StateFunc)

type WindowsOperator

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

func NewWindowsOperator

func NewWindowsOperator(expr ruleql.Expr, window *ruleql.WindowExpr,
	funcs map[string]*ruleql.CallExpr) *WindowsOperator