transporter

Documentation

Overview

Package transporter provides all adaptoremented functionality to move data through transporter.

Index

• Constants
• type Node
  • func NewNode(name, kind string, extra adaptor.Config) *Node
  • func (n *Node) Add(node *Node) *Node
  • func (n *Node) Endpoints() map[string]string
  • func (n *Node) Init(interval time.Duration) (err error)
  • func (n *Node) Path() string
  • func (n *Node) Start() error
  • func (n *Node) Stop()
  • func (n *Node) String() string
  • func (n *Node) Validate() bool
• type Pipeline
  • func NewDefaultPipeline(source *Node, uri, key, pid string, interval time.Duration) (*Pipeline, error)
  • func NewPipeline(source *Node, emitter events.Emitter, interval time.Duration) (*Pipeline, error)
  • func (pipeline *Pipeline) Run() error
  • func (pipeline *Pipeline) Stop()
  • func (pipeline *Pipeline) String() string

Constants

const (
	VERSION = "0.1.0"
)

VERSION the library

Types

type Node

type Node struct {
	Name     string         `json:"name"`     // the name of this node
	Type     string         `json:"type"`     // the node's type, used to create the adaptorementation
	Extra    adaptor.Config `json:"extra"`    // extra config options that are passed to the adaptorementation
	Children []*Node        `json:"children"` // the nodes are set up as a tree, this is an array of this nodes children
	Parent   *Node          `json:"parent"`   // this node's parent node, if this is nil, this is a 'source' node
	// contains filtered or unexported fields
}

A Node is the basic building blocks of transporter pipelines. Nodes are constructed in a tree, with the first node broadcasting data to each of it's children. Node tree's can be constructed as follows:

source := transporter.NewNode("name1", "mongo", adaptor.Config{"uri": "mongodb://localhost/boom", "namespace": "boom.foo", "debug": true})
sink1 := transporter.NewNode("foofile", "file", adaptor.Config{"uri": "stdout://"})
sink2 := transporter.NewNode("foofile2", "file", adaptor.Config{"uri": "stdout://"})
source.Add(sink1)
source.Add(sink2)

func NewNode

func NewNode(name, kind string, extra adaptor.Config) *Node

NewNode creates a new Node struct

func (*Node) Add

func (n *Node) Add(node *Node) *Node

Add the given node as a child of this node. This has side effects, and sets the parent of the given node

func (*Node) Endpoints

func (n *Node) Endpoints() map[string]string

Endpoints recurses down the node tree and accumulates a map associating node name with node type this is primarly used with the boot event

func (*Node) Init

func (n *Node) Init(interval time.Duration) (err error)

Init sets up the node for action. It creates a pipe and adaptor for this node, and then recurses down the tree calling Init on each child

func (*Node) Path

func (n *Node) Path() string

Path returns a string representation of the names of all the node's parents concatenated with "/" used in metrics eg. for the following tree source := transporter.NewNode("name1", "mongo", adaptor.Config{"uri": "mongodb://localhost/boom", "namespace": "boom.foo", "debug": true})

sink1 := transporter.NewNode("foofile", "file", adaptor.Config{"uri": "stdout://"})
source.Add(sink1)

'source' will have a Path of 'name1', and 'sink1' will have a path of 'name1/sink1'

func (*Node) Start

func (n *Node) Start() error

Start starts the nodes children in a go routine, and then runs either Start() or Listen() on the node's adaptor. Root nodes (nodes with no parent) will run Start() and will emit messages to it's children, All descendant nodes run Listen() on the adaptor

func (*Node) Stop

func (n *Node) Stop()

Stop this node's adaptor, and sends a stop to each child of this node

func (*Node) String

func (n *Node) String() string

String

func (*Node) Validate

func (n *Node) Validate() bool

Validate ensures that the node tree conforms to a proper structure. Node trees must have at least one source, and one sink. dangling transformers are forbidden. Validate only knows about default adaptors in the adaptor package, it can't validate any custom adaptors

type Pipeline

type Pipeline struct {

	// Err is the fatal error that was sent from the adaptor
	// that caused us to stop this process.  If this is nil, then
	// the transporter is running
	Err error
	// contains filtered or unexported fields
}

A Pipeline is a the end to end description of a transporter data flow. including the source, sink, and all the transformers along the way

func NewDefaultPipeline

func NewDefaultPipeline(source *Node, uri, key, pid string, interval time.Duration) (*Pipeline, error)

NewDefaultPipeline returns a new Transporter Pipeline with the given node tree, and uses the events.HttpPostEmitter to deliver metrics. eg.

  source :=
  	transporter.NewNode("source", "mongo", adaptor.Config{"uri": "mongodb://localhost/", "namespace": "boom.foo", "debug": false, "tail": true}).
	  	Add(transporter.NewNode("out", "file", adaptor.Config{"uri": "stdout://"}))
  pipeline, err := transporter.NewDefaultPipeline(source, events.Api{URI: "http://localhost/endpoint"}, 1*time.Second)
  if err != nil {
	  fmt.Println(err)
	  os.Exit(1)
  }

pipeline.Run()

func NewPipeline

func NewPipeline(source *Node, emitter events.Emitter, interval time.Duration) (*Pipeline, error)

NewPipeline creates a new Transporter Pipeline using the given tree of nodes, and Event Emitter eg.

  source :=
  	transporter.NewNode("source", "mongo", adaptor.Config{"uri": "mongodb://localhost/", "namespace": "boom.foo", "debug": false, "tail": true}).
	  	Add(transporter.NewNode("out", "file", adaptor.Config{"uri": "stdout://"}))
  pipeline, err := transporter.NewPipeline(source, events.NewNoopEmitter(), 1*time.Second)
  if err != nil {
	  fmt.Println(err)
	  os.Exit(1)
  }

pipeline.Run()

func (*Pipeline) Run

func (pipeline *Pipeline) Run() error

Run the pipeline

func (*Pipeline) Stop

func (pipeline *Pipeline) Stop()

Stop sends a stop signal to the emitter and all the nodes, whether they are running or not. the node's database adaptors are expected to clean up after themselves, and stop will block until all nodes have stopped successfully

func (*Pipeline) String

func (pipeline *Pipeline) String() string