flows

Documentation

Index

• Constants
• func NewEventDispatcher(eb baseEventSubscriber) flowEventDispatcher
• type ActionFlow
• type AllowedEvents
• type Cooldown
• type Description
• type EventFlow
  • func NewEventFlow(flow *Flow, event *ce.Event) *EventFlow
  • func (ef *EventFlow) Exec() error
• type EventFlowStatus
• type Filters
• type Flow
  • func NewFlow(manager FlowManager, cache FlowStore, id string, name string, opts ...Option) *Flow
  • func (f *Flow) Handle(event interface{})
• type FlowConfig
  • func (c FlowConfig) Validate(plugins plugin.PluginManager, id string) error
• type FlowConfigs
• type FlowManager
• type FlowStore
• type GroupBy
• type InMemoryFlowStore
  • func NewInMemFlowStore() *InMemoryFlowStore
  • func (i *InMemoryFlowStore) Delete(key string)
  • func (i *InMemoryFlowStore) Get(key string) (*EventFlow, error)
  • func (i *InMemoryFlowStore) Set(key string, ef *EventFlow, ttl time.Duration) error
• type Manager
  • func NewManager(logger log.Logger, errorHandler emperror.Handler, ...) *Manager
  • func (m *Manager) ErrorHandler() emperror.Handler
  • func (m *Manager) LoadFlows(v *viper.Viper) error
  • func (m *Manager) Logger() log.Logger
  • func (m *Manager) Plugins() plugin.PluginManager
• type Option
• type Plugins

Constants

const (
	CEIncomingTopic = "cloud.events.incoming"
)

Functions

func NewEventDispatcher

func NewEventDispatcher(eb baseEventSubscriber) flowEventDispatcher

NewEventDispatcher returns an initialized eventDispatcher

Types

type ActionFlow

type ActionFlow interface {
	Handle(event interface{})
}

ActionFlow defines an action flow

type AllowedEvents

type AllowedEvents []string

AllowedEvents defines allowed event types for the flow

type Cooldown

type Cooldown time.Duration

Cooldown time that passes after an event flow is successfully finished

type Description

type Description string

Description sets the description of the action flow

type EventFlow

type EventFlow struct {
	Status EventFlowStatus
	Error  error
	// contains filtered or unexported fields
}

EventFlow is an actual sequential executing of defined plugins for a particular event and a defined action flow

func NewEventFlow

func NewEventFlow(flow *Flow, event *ce.Event) *EventFlow

NewEventFlow returns an initialized EventFlow

func (*EventFlow) Exec

func (ef *EventFlow) Exec() error

Exec executes the defined plugins sequentially

type EventFlowStatus

type EventFlowStatus string

const (
	EventFlowCompleted   EventFlowStatus = "completed"
	EventFlowFailed      EventFlowStatus = "failed"
	EventFlowInProgress  EventFlowStatus = "inprogress"
	EventFlowInitialized EventFlowStatus = "initialized"
	EventFlowCoolingDown EventFlowStatus = "coolingdown"
)

type Filters

type Filters map[string]string

Filters defines simple filter on event values

type Flow

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

Flow describes an action flow

func NewFlow

func NewFlow(manager FlowManager, cache FlowStore, id string, name string, opts ...Option) *Flow

NewFlow returns an initialized action flow

func (*Flow) Handle

func (f *Flow) Handle(event interface{})

Handle handles the event by starting and event flow which executes the defined plugins

type FlowConfig

type FlowConfig struct {
	Name    string   `mapstructure:"name"`
	Plugins []string `mapstructure:"plugins"`

	Description   string            `mapstructure:"description"`
	AllowedEvents []string          `mapstructure:"allowedEvents"`
	GroupBy       []string          `mapstructure:"groupBy"`
	Filters       map[string]string `mapstructure:"filters"`
	Cooldown      time.Duration     `mapstructure:"cooldown"`
}

FlowConfig holds configuration values for an action flow

func (FlowConfig) Validate

func (c FlowConfig) Validate(plugins plugin.PluginManager, id string) error

Validate validates flow configuration

type FlowConfigs

type FlowConfigs map[string]FlowConfig

type FlowManager

type FlowManager interface {
	Logger() log.Logger
	ErrorHandler() emperror.Handler
	Plugins() plugin.PluginManager
}

FlowManager is used for managing action flows

type FlowStore

type FlowStore interface {
	Get(string) (*EventFlow, error)
	Set(string, *EventFlow, time.Duration) error
	Delete(string)
}

type GroupBy

type GroupBy []string

GroupBy categorizes subsequent events as the same if all the corresponding values of these attributes match

type InMemoryFlowStore

type InMemoryFlowStore struct {
	EventFlowCache *cache.Cache
}

func NewInMemFlowStore

func NewInMemFlowStore() *InMemoryFlowStore

func (*InMemoryFlowStore) Delete

func (i *InMemoryFlowStore) Delete(key string)

func (*InMemoryFlowStore) Get

func (i *InMemoryFlowStore) Get(key string) (*EventFlow, error)

func (*InMemoryFlowStore) Set

func (i *InMemoryFlowStore) Set(key string, ef *EventFlow, ttl time.Duration) error

type Manager

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

Manager describes a FlowManager implementation

func NewManager

func NewManager(logger log.Logger, errorHandler emperror.Handler, dispatcher flowEventDispatcher, plugins plugin.PluginManager) *Manager

NewManager returns an initialized FlowManager implementation

func (*Manager) ErrorHandler

func (m *Manager) ErrorHandler() emperror.Handler

ErrorHandler returns the error handler

func (*Manager) LoadFlows

func (m *Manager) LoadFlows(v *viper.Viper) error

LoadFlows loads flow definitions from config, initializes Flows and subscribes them to the event dispatcher

func (*Manager) Logger

func (m *Manager) Logger() log.Logger

Logger returns the logger

func (*Manager) Plugins

func (m *Manager) Plugins() plugin.PluginManager

Plugins returns the plugin manager

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option sets configuration on the Flow

type Plugins

type Plugins []string

Plugins defines the plugins to execute in an event flow