node

Documentation

Index

• type AsyncNode
• type Config
• type Context
  • func InitNodeContextHierarchy(nodeConfig *Config) *Context
  • func (nc *Context) ProcessEvent(event *firebolt.Event)
• type Node
• type Registration
• type Registry
  • func GetRegistry() *Registry
  • func (r *Registry) GetNodeRegistration(nodeType string) *Registration
  • func (r *Registry) GetSourceRegistration(sourceType string) *SourceRegistration
  • func (r *Registry) InstantiateNode(nodeType string) Node
  • func (r *Registry) InstantiateSource(sourceType string) Source
  • func (r *Registry) RegisterNodeType(nodeType string, factory func() Node, consumes reflect.Type, ...)
  • func (r *Registry) RegisterSourceType(sourceType string, factory func() Source, produces reflect.Type)
• type Source
• type SourceConfig
• type SourceRegistration
• type SyncNode

Types

type AsyncNode

type AsyncNode interface {
	Node
	ProcessAsync(event *firebolt.AsyncEvent)
}

AsyncNode is a Node which responds asynchronously.

type Config

type Config struct {
	ID                  string            `yaml:"id"`
	Name                string            `yaml:"name"`
	Workers             int               `yaml:"workers"`
	BufferSize          int               `yaml:"buffersize"`
	Params              map[string]string `yaml:"params"`
	Children            []*Config         `yaml:"children"`
	ErrorHandler        *Config           `yaml:"error_handler"`
	Disabled            bool              `yaml:"disabled"`
	DiscardOnFullBuffer bool              `yaml:"discard_on_full_buffer"`
}

Config is the set of configuration values used to initialize the Context for a Node

type Context

type Context struct {
	Config        *Config
	Async         bool
	Ch            chan firebolt.Event // ch is the input buffer for this node
	StopCh        chan bool
	NodeProcessor Node
	Children      []*Context
	ErrorHandler  *Context
	MessageTypes  []string
	WaitGroup     *sync.WaitGroup
	ShutdownOnce  *sync.Once
}

Context is the execution context for a given node

func InitNodeContextHierarchy

func InitNodeContextHierarchy(nodeConfig *Config) *Context

InitNodeContextHierarchy recursively builds the hierarchy of node execution Contexts starting at the passed node.

func (*Context) ProcessEvent

func (nc *Context) ProcessEvent(event *firebolt.Event)

ProcessEvent calls the node's 'process' method for the passed event and handles success / filter / failure cases.

type Node

type Node interface {
	Init(id string, ctx fbcontext.FBContext)
	AcceptsMessage(messageType string) bool
	Setup(config map[string]string) error
	Shutdown() error
	Receive(msg fbcontext.Message) error
}

Node is a single unit of processing; a tree of Nodes comprise all event processing in a user application.

type Registration

type Registration struct {
	Consumes reflect.Type
	Produces reflect.Type
	// contains filtered or unexported fields
}

Registration is the record representing a node type and metadata about that node

type Registry

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

Registry is the singleton containing all source and node types registered in firebolt and therefore available to use in your `firebolt.yaml` configuration. Built-in types can be used immediately, while any custom sources or nodes that your application provides will need to be registered with `registry.Get().RegisterSourceType` or `registry.Get().RegisterNodeType` respectively.

func GetRegistry

func GetRegistry() *Registry

GetRegistry returns the singleton instance of the source/node registry. The registry is lazily initialized on the first invocation.

func (*Registry) GetNodeRegistration

func (r *Registry) GetNodeRegistration(nodeType string) *Registration

GetNodeRegistration returns the NodeRegistration associated with a node type name, or nil if no match exists

func (*Registry) GetSourceRegistration

func (r *Registry) GetSourceRegistration(sourceType string) *SourceRegistration

GetSourceRegistration returns the SourceRegistration associated with a source type name, or nil if no registration exists

func (*Registry) InstantiateNode

func (r *Registry) InstantiateNode(nodeType string) Node

InstantiateNode creates and returns a Node instance of the passed type. This method will panic in the event that the node type does not exist.

func (*Registry) InstantiateSource

func (r *Registry) InstantiateSource(sourceType string) Source

InstantiateSource creates and returns a Source instance of the passed type. This method will panic in the event that the source type does not exist.

func (*Registry) RegisterNodeType

func (r *Registry) RegisterNodeType(nodeType string, factory func() Node, consumes reflect.Type, produces reflect.Type)

RegisterNodeType registers a new node type on behalf of application code; all node types referenced in the configuration must be registered before the executor is invoked.

func (*Registry) RegisterSourceType

func (r *Registry) RegisterSourceType(sourceType string, factory func() Source, produces reflect.Type)

RegisterSourceType registers a new source type on behalf of application code; the source type referenced in the configuration must be registered before the executor is invoked.

type Source

type Source interface {
	Init(id string, ctx fbcontext.FBContext)
	AcceptsMessage(messageType string) bool
	Setup(config map[string]string, eventchan chan firebolt.Event) error
	Start() error
	Shutdown() error
	Receive(msg fbcontext.Message) error
}

Source is a root node that generates messages to be processed or receives them from an external system.

type SourceConfig

type SourceConfig struct {
	Name   string            `yaml:"name"`
	ID     string            `yaml:"id"`
	Params map[string]string `yaml:"params"`
}

SourceConfig is the set of configuration values for this Source

type SourceRegistration

type SourceRegistration struct {
	Produces reflect.Type
	// contains filtered or unexported fields
}

SourceRegistration is the record representing a source type and metadata about that source

type SyncNode

type SyncNode interface {
	Node
	Process(event *firebolt.Event) (*firebolt.Event, error)
}

SyncNode is a Node which responds synchronously.