flo

README

flo

A dataflow like library for Go.

Introduction

Graphs are processing pipelines, you can think of them as a template or class, a graph can be started multiple times with a different name and configuration. Below each ... represents a function that the user of flo supplies.

g := graph.New()
g.From(...)
g.Transform(...)
g.Group(...)
g.Window(...)
g.Trigger(...)
g.Into(...)

Graphs get their data from sources, and write their outputs into sinks. Both sources and sinks are setup respective to the graph type and name they are being used.

type Sources interface {
	Setup(graphType, graphName, conf) []Source
}

type Sinks interface {
	Setup(graphType, graphName, conf) []Sink
}

The process is initialized by registering the messages used for processing, which must be Protobuf messages, and registering the graphs used for processing.

flo.RegisterMsg(...)
flo.RegisterGraph(...)

Since flo is just a library, processing is done the Go way, by building a static binary which just uses the flo server from its main function.

server, ... := flo.NewServer(...)
server.Serve(...)

Once the server is running, a graph is started using the flo client from anywhere. The client writes an entry to etcd for each graph instance. The server, watching etcd events, responds to the client.

client, ... := flo.NewClient(...)
client.StartGraph(graphType, graphName, conf)

Documentation

Index

• Variables
• func LookupGraph(graphType string) (*graph.Definition, bool)
• func RegisterGraph(graphType string, g *graph.Graph) error
• func RegisterMsg(v interface{}) error
• type Cfg
• type Client
  • func NewClient(etcd *etcdv3.Client, cfg Cfg) (*Client, error)
  • func (c *Client) RunGraph(graphType, graphName string, config []byte) error
  • func (c *Client) TerminateGraph(graphType, graphName string) error
  • func (c *Client) Watch(ctx context.Context) ([]*registry.WatchEvent, <-chan *registry.WatchEvent, error)
• type Server
  • func NewServer(etcd *etcdv3.Client, cfg Cfg) (*Server, error)
  • func (s *Server) Serve(lis net.Listener) error
  • func (s *Server) Stop()

Variables

var (
	// ErrNilGraph when a nil graph is defined.
	ErrNilGraph = errors.New("nil graph")
	// ErrInvalidStorage when no storage configuration is given, or
	// something within the configuration is invalid.
	ErrInvalidStorage = errors.New("invalid storage")
	// ErrAlreadyDefined when a graph is defined more than once.
	ErrAlreadyDefined = errors.New("already defined")
	// ErrInvalidGraphType when the graph type contains invalid
	// characters or is the empty string.
	ErrInvalidGraphType = errors.New("invalid graph type")
	// ErrInvalidNamespace when invalid characters appear in the namespace
	// or the namespace is the empty string.
	ErrInvalidNamespace = errors.New("invalid namespace")
)

Functions

func LookupGraph

func LookupGraph(graphType string) (*graph.Definition, bool)

LookupGraph definition that was previously registered.

func RegisterGraph

func RegisterGraph(graphType string, g *graph.Graph) error

RegisterGraph of the given graph type.

func RegisterMsg

func RegisterMsg(v interface{}) error

RegisterMsg where v is a non-pointer protobuf message type.

Types

type Cfg

type Cfg struct {
	Driver    driver.Cfg
	Namespace string
}

Cfg where the only required parameter is the namespace.

type Client

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

Client for graphs.

func NewClient

func NewClient(etcd *etcdv3.Client, cfg Cfg) (*Client, error)

NewClient to start and stop graphs.

func (*Client) RunGraph

func (c *Client) RunGraph(graphType, graphName string, config []byte) error

RunGraph of the given type and name.

func (*Client) TerminateGraph

func (c *Client) TerminateGraph(graphType, graphName string) error

TerminateGraph of the given type and name.

func (*Client) Watch

func (c *Client) Watch(ctx context.Context) ([]*registry.WatchEvent, <-chan *registry.WatchEvent, error)

Watch for graph registration and other lifecycle events.

type Server

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

func NewServer

func NewServer(etcd *etcdv3.Client, cfg Cfg) (*Server, error)

func (*Server) Serve

func (s *Server) Serve(lis net.Listener) error

func (*Server) Stop

func (s *Server) Stop()