types

Documentation

Index

• Constants
• Variables
• type Action
• type ActionBuilder
• type ActionContent
• type BaseObject
• type CallbackFunc
• type DBImpl
• type EventActionType
• type EventType
• type FlushAction
• type GrpcClient
• type JobManagerClient
• type Message
• type MsgSlice
• type PubSubClient
• type PublishMessage
• type ResourceCallbackFunc
• type ResourceEvent
• type ResourceEventValue
• type ResourceManager
• type ResourcesManagerClient
• type RuleActionClient
• type STATUS
• type ServiceConfig
• type SinkEvent
  • func (*SinkEvent) EventType()
• type Slot
• type SourceEvent
  • func (*SourceEvent) EventType()
• type SyncManager
• type Task
• type TaskEvent
  • func (*TaskEvent) EventType()
• type TaskGroup
• type TaskID
• type Type

Constants

const (
	LoopbackStream = "loopback://memqueue/republish/router"

	MATE_RULE_ID   = "x-rule-id"
	MATE_RULE_BODY = "x-rule-body"
)

const (
	RuleEvent = EventType(iota)
	PubSubEvent
	RouteEvent
)

const (
	PUT = EventActionType(iota)
	DELETE
)

Variables

var (
	ErrMetadataNotReachable = fmt.Errorf("metadata service not reachable")
	ErrCreatTaskFail        = fmt.Errorf("creat task fail")
	ErrDeployTaskFail       = fmt.Errorf("deploy task fail")
	ErrRecvUnknownType      = errors.New("recv error(unknown type)")

	ErrFilterFalse     = errors.New("rule eval filter false")
	ErrTaskUserIDEmpty = errors.New("task userID empty")
	ErrOptionParseFail = errors.New("parse option empty")

	ErrRestore = fmt.Errorf("metadata restore error")
	ErrSync    = fmt.Errorf("metadata sync error")

	ErrDecode = fmt.Errorf("message decode fail")
)

var (
	IDLE     = STATUS(0)
	STARTING = STATUS(1)
	STARTED  = STATUS(2)
	ERROR    = STATUS(-1)
)

var DirectEntityKey struct{}

Direct Pubsub Entity Key

var NewMessage = pb.NewMessage

var TypeOf = reflect.TypeOf

Types

type Action

type Action interface {
	Setup(ctx ActionContent, metadata map[string]string) error
	Invoke(ctx ActionContent, messages v1.Message) error
	Close() error
}

type ActionBuilder

type ActionBuilder func(entityType, entityID string) Action

type ActionContent

type ActionContent interface {
	Ack()
	Nack(error)
	Context() context.Context
}

type BaseObject

type BaseObject interface {
	ID() string
	UserID() string
	String() string
	RefreshTime() int64
}

Object

type CallbackFunc

type CallbackFunc func(ctx context.Context, evt interface{}) error

type DBImpl

type DBImpl interface {
	BatchCreate(ctx context.Context, sqlCreate string, values interface{}) error
	Close() error
}

type EventActionType

type EventActionType int32

type EventType

type EventType int32

type FlushAction

type FlushAction interface {
	Action
	Flush(ctx ActionContent) error
}

type GrpcClient

type GrpcClient interface {
	Manager() pb.JobManagerClient
	PubSub() pb.PubSubClient
	Rule() pb.RuleActionClient
	Resource() pb.ResourcesManagerClient
}

type JobManagerClient

type JobManagerClient = pb.JobManagerClient

type Message

type Message = pb.Message

type MsgSlice

type MsgSlice = []stream.PublishMessage

type PubSubClient

type PubSubClient = pb.PubSubClient

type PublishMessage

type PublishMessage = pb.PublishMessage

type ResourceCallbackFunc

type ResourceCallbackFunc func(ctx context.Context, evt *metapb.ResourceObject) error

type ResourceEvent

type ResourceEvent interface {
	EventType()
}

type ResourceEventValue

type ResourceEventValue struct {
	Type metapb.ResourceObject_EventType
	Body interface{}
}

type ResourceManager

type ResourceManager interface {
	Sink(dsn string) (sink stream.Sink, err error)
	Sources() map[string]stream.Source
	RePublishSink() stream.Sink
	PubsubSink() stream.Sink
	SetPubsubSink(stream.Sink)
	Slot(userID string) Slot
	HandleEvent(ctx context.Context, evt *metapb.ResourceObject) error
}

type ResourcesManagerClient

type ResourcesManagerClient = pb.ResourcesManagerClient

type RuleActionClient

type RuleActionClient = pb.RuleActionClient

type STATUS

type STATUS int

type ServiceConfig

type ServiceConfig struct {
	Client GrpcClient
	Logger log.Factory
	Tracer tracing.Tracer
}

type SinkEvent

type SinkEvent struct {
	Stream string
}

func (*SinkEvent) EventType

func (*SinkEvent) EventType()

type Slot

type Slot interface {
	AddTask(ctx context.Context, event *TaskEvent) bool
	DelTask(ctx context.Context, event *TaskEvent) bool
	Invoke(ctx context.Context, evalCtx ruleql.Context, messages stream.PublishMessage) error
	Tree() *topic.Tree
}

Slot User slot with private tire topic tree

type SourceEvent

type SourceEvent struct {
	Stream string
}

func (*SourceEvent) EventType

func (*SourceEvent) EventType()

type SyncManager

type SyncManager interface {
	Run()
	SetReceiver(ctx context.Context, fn ResourceCallbackFunc)
}

type Task

type Task interface {
	BaseObject
	Invoke(ctx context.Context, evalCtx ruleql.Context, msg stream.PublishMessage) error
	Close(ctx context.Context) error
}

Task

type TaskEvent

type TaskEvent struct {
	Type   EventType
	UserId string
	Topic  string
	Task   Task
}

func (*TaskEvent) EventType

func (*TaskEvent) EventType()

type TaskGroup

type TaskGroup = []Task

type TaskID

type TaskID int32

type Type

type Type = reflect.Type