pipe

Documentation

Overview

Package pipe provides types to help manage transporter communication channels as well as event types.

Index

• type Pipe
  • func NewPipe(pipe *Pipe, path string) *Pipe
  • func (m *Pipe) Listen(fn func(*message.Msg) (*message.Msg, error), nsFilter *regexp.Regexp) error
  • func (m *Pipe) Send(msg *message.Msg)
  • func (m *Pipe) Stop()

Types

type Pipe

type Pipe struct {
	In      messageChan
	Out     []messageChan
	Err     chan error
	Event   chan events.Event
	Stopped bool // has the pipe been stopped?

	MessageCount int
	LastMsg      *message.Msg
	ExtraState   map[string]interface{}
	// contains filtered or unexported fields
}

Pipe provides a set of methods to let transporter nodes communicate with each other.

Pipes contain In, Out, Err, and Event channels. Messages are consumed by a node through the 'in' chan, emited from the node by the 'out' chan. Pipes come in three flavours, a sourcePipe, which only emits messages and has no listening loop, a sinkPipe which has a listening loop, but doesn't emit any messages, and joinPipe which has a li tening loop that also emits messages.

func NewPipe

func NewPipe(pipe *Pipe, path string) *Pipe

NewPipe creates a new Pipe. If the pipe that is passed in is nil, then this pipe will be treaded as a source pipe that just serves to emit messages. Otherwise, the pipe returned will be created and chained from the last member of the Out slice of the parent. This function has side effects, and will add an Out channel to the pipe that is passed in

func (*Pipe) Listen

func (m *Pipe) Listen(fn func(*message.Msg) (*message.Msg, error), nsFilter *regexp.Regexp) error

Listen starts a listening loop that pulls messages from the In chan, applies fn(msg), a `func(message.Msg) error`, and emits them on the Out channel. Errors will be emited to the Pipe's Err chan, and will terminate the loop. The listening loop can be interupted by calls to Stop().

func (*Pipe) Send

func (m *Pipe) Send(msg *message.Msg)

Send emits the given message on the 'Out' channel. the send Timesout after 100 ms in order to chaeck of the Pipe has stopped and we've been asked to exit. If the Pipe has been stopped, the send will fail and there is no guarantee of either success or failure

func (*Pipe) Stop

func (m *Pipe) Stop()

Stop terminates the channels listening loop, and allows any timeouts in send to fail