ekuiper-new/xstream/api/stream.go

package api

import (
	"context"
	"github.com/sirupsen/logrus"
)

//The function to call when data is emitted by the source.
type ConsumeFunc func(message map[string]interface{}, metadata map[string]interface{})

type Closable interface {
	Close(ctx StreamContext) error
}

type Source interface {
	//Should be sync function for normal case. The container will run it in go func
	Open(ctx StreamContext, consume ConsumeFunc) error
	//Called during initialization. Configure the source with the data source(e.g. topic for mqtt) and the properties
	//read from the yaml
	Configure(datasource string, props map[string]interface{}) error
	Closable
}

type Sink interface {
	//Should be sync function for normal case. The container will run it in go func
	Open(ctx StreamContext) error
	//Called during initialization. Configure the sink with the properties from rule action definition
	Configure(props map[string]interface{}) error
	//Called when each row of data has transferred to this sink
	Collect(ctx StreamContext, data interface{}) error
	Closable
}

type Emitter interface {
	AddOutput(chan<- interface{}, string) error
}

type Collector interface {
	GetInput() (chan<- interface{}, string)
}

type TopNode interface {
	GetName() string
}

type Rule struct {
	Id      string                   `json:"id"`
	Sql     string                   `json:"sql"`
	Actions []map[string]interface{} `json:"actions"`
	Options map[string]interface{}   `json:"options"`
}

type StreamContext interface {
	context.Context
	GetLogger()  *logrus.Entry
	GetRuleId() string
	GetOpId() string
	WithMeta(ruleId string, opId string) StreamContext
	WithCancel() (StreamContext, context.CancelFunc)
}

type Operator interface {
	Emitter
	Collector
	Exec(StreamContext, chan<- error)
	GetName() string
}

type Function interface {
	//The argument is a list of xsql.Expr
	Validate(args []interface{}) error
	//Execute the function, return the result and if execution is successful.
	//If execution fails, return the error and false.
	Exec(args []interface{}) (interface{}, bool)
}