storm

README

storm

Documentation

Index

• Constants
• type Bolt
  • func NewBolt(s *Storm, p BoltProcessor) *Bolt
  • func (b *Bolt) Run()
• type BoltConfiguration
  • func (bc *BoltConfiguration) Input(source EmiterName, distributeHow DistributeHow, fields []string)
  • func (bc *BoltConfiguration) Name() string
• type BoltInput
• type BoltOutput
• type BoltProcessor
• type ComponentDef
• type ConnectInfo
• type DistributeHow
• type EmiterName
• type ProcessId
• type Spout
  • func NewSpout(s *Storm, spout Spouter) *Spout
  • func (s *Spout) Run()
• type SpoutMessage
• type Spouter
• type Storm
  • func NewStorm() *Storm
  • func (s *Storm) End()
  • func (s *Storm) Run()
• type Topology
  • func NewTopology(name string) *Topology
  • func (t *Topology) AddBolt(name string, bolt *ComponentDef) *BoltConfiguration
  • func (t *Topology) AddSpout(name string, spout *ComponentDef) EmiterName
• type TupleMessage

Constants

const (
	DistributeByShuffle      = "shuffle"
	DistributeByField        = "field"
	DistributeToAll          = "all"
	DistributeDirect         = "direct"
	DistributeAny            = "none" //none means no choice, don't care
	DistributeLocalOrShuffle = "localorshuffle"
)

Types

type Bolt

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

Bolt - Applies a transformation on an input tuple Input Types (2 different): * Tuple (BoltInput) * TaskIds (from a previous emit)

Output: (BoltOutput) * Tupel (now changed in some way) * Command - ack, fail, log, emit

func NewBolt

func NewBolt(s *Storm, p BoltProcessor) *Bolt

NewBold - Creates a new Bolt for transformations

func (*Bolt) Run

func (b *Bolt) Run()

Process - Process all tuples that come into the bold

type BoltConfiguration

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

func (*BoltConfiguration) Input

func (bc *BoltConfiguration) Input(source EmiterName, distributeHow DistributeHow, fields []string)

func (*BoltConfiguration) Name

func (bc *BoltConfiguration) Name() string

Emiter interface implementation

type BoltInput

type BoltInput struct {
	//TupleMessage - the tuple and it's metadata
	TupleMessage

	//Component - The id of the creating component
	Component string `json:"comp"`
}

BoltInput - Inbound Tuple

type BoltOutput

type BoltOutput struct {

	//TupleMessage - the tuple and it's metadata
	TupleMessage

	//Anchors - The ids of the tuples these output tuples should be anchored to
	Anchors []string `json:"anchors"`

	//Command - ack, fail, emit, log
	Command string `json:"command"`

	//Message - for log only
	Message string `json:"message",omitempty`
}

BoltInput - Outbound Tuple

type BoltProcessor

type BoltProcessor interface {
	Process(tuple *TupleMessage) (*TupleMessage, error)
	TrackIndirectEmit(taskIds []int)
}

Transformer - Enables an implementation of a Bolt Transform

type ComponentDef

type ComponentDef struct {
	ShellCommand string //to execute
	OutputFields []string
	Direct       bool
	Parallelism  int32
}

type ConnectInfo

type ConnectInfo struct {
	Conf    *json.RawMessage `json:"conf"`
	Context *json.RawMessage `json:"context"`
	PidDir  string           `json:"pidDir"`
}

type DistributeHow

type DistributeHow string

type EmiterName

type EmiterName interface {
	Name() string
}

type ProcessId

type ProcessId struct {
	Pid string `json:"pid"`
}

type Spout

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

Spout - Emits data Input: (Message)

func NewSpout

func NewSpout(s *Storm, spout Spouter) *Spout

NewSpout - Creates a new spout for the given storm session.

func (*Spout) Run

func (s *Spout) Run()

Run - Runs the spout

type SpoutMessage

type SpoutMessage struct {

	//TupleMessage - the tuple and it's metadata
	TupleMessage

	// Command - next, ack, fail, emit, log, sync
	Command string `json:"command"`

	//Message - for log only
	Message string `json:"message",omitempty`
}

Spout Input & Output

type Spouter

type Spouter interface {
	Emit() *TupleMessage
	Ack(id string)
	Fail(id string)
	AssociateTasks(id string, taskIds []int)
}

type Storm

type Storm struct {
	Input  chan []byte
	Output chan interface{}

	ExtIn  io.Reader
	ExtOut io.Writer
	// contains filtered or unexported fields
}

Storm - The storm Processor for running a Spout or a Bolt

func NewStorm

func NewStorm() *Storm

NewStormSession - Connects with Storm and starts the processor for running a Bolt or Spout

func (*Storm) End

func (s *Storm) End()

func (*Storm) Run

func (s *Storm) Run()

type Topology

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

func NewTopology

func NewTopology(name string) *Topology

func (*Topology) AddBolt

func (t *Topology) AddBolt(name string, bolt *ComponentDef) *BoltConfiguration

func (*Topology) AddSpout

func (t *Topology) AddSpout(name string, spout *ComponentDef) EmiterName

type TupleMessage

type TupleMessage struct {

	//Id - To identifiy this tuple, for messaging guarantees
	Id string `json:"id",omitempty`

	//Stream - Id of the stream this tuple is emmited to. Blank==default stream
	Stream string `json:"stream",omitempty`

	//Task - For direct emit
	Task *int32 `json:"task",omitempty`

	//Tuple - A 'row' of data
	Tuple []interface{} `json:"tuple"`
}

TupleMessage - A tuple an it's processing metadata