pipelines

README

Pipelines

Cloud based DAG infrastructure with inter-node release.

POC Instructions

$ gnatsd -m 8222
$ nats-top
$ export PATH=$PATH:$GOPATH/bin
$ go install ./...
$ pipeline server || pipeline server nats://localhsot:4222 --test
$ pipeline agent  # As many agents as you would like
$ pipeline load sample/web
$ pipeline send crawl_request https://en.wikipedia.org/wiki/Main_page

Documentation

Overview

Package pipelines is a generated protocol buffer package.

It is generated from these files:

pipelines.proto

It has these top-level messages:

Emit
Work
Record
Timer
StartWorker

Index

• Variables
• func EmitRecord(stream string, record *Record) error
• func Register(name string, comp Computation)
• func Run()
• type Computation
• type Emit
  • func (*Emit) Descriptor() ([]byte, []int)
  • func (m *Emit) GetRecord() *Record
  • func (*Emit) ProtoMessage()
  • func (m *Emit) Reset()
  • func (m *Emit) String() string
• type Record
  • func NewRecord(data string) *Record
  • func (r Record) AsTest() *Record
  • func (*Record) Descriptor() ([]byte, []int)
  • func (r *Record) New(data string) *Record
  • func (*Record) ProtoMessage()
  • func (m *Record) Reset()
  • func (m *Record) String() string
• type StartWorker
  • func (*StartWorker) Descriptor() ([]byte, []int)
  • func (*StartWorker) ProtoMessage()
  • func (m *StartWorker) Reset()
  • func (m *StartWorker) String() string
• type Timer
  • func (*Timer) Descriptor() ([]byte, []int)
  • func (*Timer) ProtoMessage()
  • func (m *Timer) Reset()
  • func (m *Timer) String() string
• type Work
  • func (*Work) Descriptor() ([]byte, []int)
  • func (m *Work) GetRecord() *Record
  • func (*Work) ProtoMessage()
  • func (m *Work) Reset()
  • func (w Work) ServiceKey() string
  • func (m *Work) String() string

Variables

var ErrKillMeNow = errors.New("KILL ME NOW")

ErrKillMeNow is used for now to allow the clients to forcaviley kill themselves

Functions

func EmitRecord

func EmitRecord(stream string, record *Record) error

EmitRecord transmits a record to the system

func Register

func Register(name string, comp Computation)

Register registers a parent instance of a computaton as a potential worker

func Run

func Run()

Run starts the entire node

Types

type Computation

type Computation interface {
	Start(context.Context, func()) (context.Context, error)
	ProcessRecord(*Record) error
	ProcessTimer(*Timer) error
}

Computation is the base interface for all working operations

type Emit

type Emit struct {
	Record *Record `protobuf:"bytes,1,opt,name=record" json:"record,omitempty"`
	Stream string  `protobuf:"bytes,2,opt,name=stream" json:"stream,omitempty"`
}

func (*Emit) Descriptor

func (*Emit) Descriptor() ([]byte, []int)

func (*Emit) GetRecord

func (m *Emit) GetRecord() *Record

func (*Emit) ProtoMessage

func (*Emit) ProtoMessage()

func (*Emit) Reset

func (m *Emit) Reset()

func (*Emit) String

func (m *Emit) String() string

type Record

type Record struct {
	CorrelationID uint64 `protobuf:"varint,1,opt,name=correlationID" json:"correlationID,omitempty"`
	Guid          uint64 `protobuf:"varint,2,opt,name=guid" json:"guid,omitempty"`
	Data          string `protobuf:"bytes,3,opt,name=data" json:"data,omitempty"`
	Test          bool   `protobuf:"varint,4,opt,name=test" json:"test,omitempty"`
}

func NewRecord

func NewRecord(data string) *Record

NewRecord constructs a completely new record

func (Record) AsTest

func (r Record) AsTest() *Record

AsTest constructs a copy of this record intended for test routing

func (*Record) Descriptor

func (*Record) Descriptor() ([]byte, []int)

func (*Record) New

func (r *Record) New(data string) *Record

New constructs new record based on a source record

func (*Record) ProtoMessage

func (*Record) ProtoMessage()

func (*Record) Reset

func (m *Record) Reset()

func (*Record) String

func (m *Record) String() string

type StartWorker

type StartWorker struct {
	Service string `protobuf:"bytes,1,opt,name=service" json:"service,omitempty"`
	Key     string `protobuf:"bytes,2,opt,name=key" json:"key,omitempty"`
	Command string `protobuf:"bytes,3,opt,name=command" json:"command,omitempty"`
	Guid    string `protobuf:"bytes,4,opt,name=guid" json:"guid,omitempty"`
}

func (*StartWorker) Descriptor

func (*StartWorker) Descriptor() ([]byte, []int)

func (*StartWorker) ProtoMessage

func (*StartWorker) ProtoMessage()

func (*StartWorker) Reset

func (m *StartWorker) Reset()

func (*StartWorker) String

func (m *StartWorker) String() string

type Timer

type Timer struct {
	CorrelationID uint64 `protobuf:"varint,1,opt,name=correlationID" json:"correlationID,omitempty"`
	Guid          uint64 `protobuf:"varint,2,opt,name=guid" json:"guid,omitempty"`
	Stamp         uint64 `protobuf:"varint,3,opt,name=stamp" json:"stamp,omitempty"`
}

func (*Timer) Descriptor

func (*Timer) Descriptor() ([]byte, []int)

func (*Timer) ProtoMessage

func (*Timer) ProtoMessage()

func (*Timer) Reset

func (m *Timer) Reset()

func (*Timer) String

func (m *Timer) String() string

type Work

type Work struct {
	Record  *Record `protobuf:"bytes,1,opt,name=record" json:"record,omitempty"`
	Service string  `protobuf:"bytes,2,opt,name=service" json:"service,omitempty"`
	Key     string  `protobuf:"bytes,3,opt,name=key" json:"key,omitempty"`
}

func (*Work) Descriptor

func (*Work) Descriptor() ([]byte, []int)

func (*Work) GetRecord

func (m *Work) GetRecord() *Record

func (*Work) ProtoMessage

func (*Work) ProtoMessage()

func (*Work) Reset

func (m *Work) Reset()

func (Work) ServiceKey

func (w Work) ServiceKey() string

ServiceKey generates a worker address for the worker designed to execute this work

func (*Work) String

func (m *Work) String() string