miniplumber

Documentation

Overview

miniplumber is a package to facilitate communication pipelines between registered readers and writers across a distributed meshage environment. miniplumber supports plumbing to external programs over stdio, similar to unix pipelines, supports trees of communication pipelines, and configurable delivery options.

Index

• Constants
• type Message
• type Pipe
  • func (p *Pipe) GetVia() string
  • func (p *Pipe) Last() string
  • func (p *Pipe) Log(mode bool)
  • func (p *Pipe) Mode() string
  • func (p *Pipe) Name() string
  • func (p *Pipe) NumMessages() int64
  • func (p *Pipe) NumReaders() int
  • func (p *Pipe) NumWriters() int
• type Plumber
  • func New(n *meshage.Node) *Plumber
  • func (p *Plumber) Log(pipe string, mode bool)
  • func (p *Plumber) Mode(pipe string, mode int)
  • func (p *Plumber) NewReader(pipe string) *Reader
  • func (p *Plumber) NewWriter(pipe string) chan<- string
  • func (p *Plumber) PipeDelete(pipe string) error
  • func (p *Plumber) PipeDeleteAll() error
  • func (p *Plumber) PipelineDelete(production ...string) error
  • func (p *Plumber) PipelineDeleteAll() error
  • func (p *Plumber) Pipelines() []string
  • func (p *Plumber) Pipes() []*Pipe
  • func (p *Plumber) Plumb(production ...string) error
  • func (p *Plumber) Via(pipe string, command []string)
  • func (p *Plumber) Write(pipe string, value string)
• type Reader
  • func (r *Reader) Close()

Constants

const (
	TIMEOUT   = time.Duration(10 * time.Second)
	TOKEN_MAX = 1024 * 1024
)

const (
	MODE_ALL = iota
	MODE_RR
	MODE_RND
)

const (
	MESSAGE_FORWARD = iota
	MESSAGE_QUERY
	MESSAGE_QUERY_RESPONSE
	MESSAGE_VIA_WRITE
	MESSAGE_VIA_HOST
)

const (
	SCHEDULE_ALL = -1
)

Types

type Message

type Message struct {
	TID     int64
	From    string
	Type    int
	Pipe    string
	Data    map[int64]string
	Readers []int64
	Value   string
}

type Pipe

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

func (*Pipe) GetVia

func (p *Pipe) GetVia() string

func (*Pipe) Last

func (p *Pipe) Last() string

func (*Pipe) Log

func (p *Pipe) Log(mode bool)

func (*Pipe) Mode

func (p *Pipe) Mode() string

func (*Pipe) Name

func (p *Pipe) Name() string

func (*Pipe) NumMessages

func (p *Pipe) NumMessages() int64

func (*Pipe) NumReaders

func (p *Pipe) NumReaders() int

func (*Pipe) NumWriters

func (p *Pipe) NumWriters() int

type Plumber

type Plumber struct {
	Messages chan *meshage.Message // incoming messages from meshage
	// contains filtered or unexported fields
}

func New

func New(n *meshage.Node) *Plumber

New returns a new Plumber object over meshage node n

func (*Plumber) Log

func (p *Plumber) Log(pipe string, mode bool)

func (*Plumber) Mode

func (p *Plumber) Mode(pipe string, mode int)

func (*Plumber) NewReader

func (p *Plumber) NewReader(pipe string) *Reader

func (*Plumber) NewWriter

func (p *Plumber) NewWriter(pipe string) chan<- string

func (*Plumber) PipeDelete

func (p *Plumber) PipeDelete(pipe string) error

func (*Plumber) PipeDeleteAll

func (p *Plumber) PipeDeleteAll() error

func (*Plumber) PipelineDelete

func (p *Plumber) PipelineDelete(production ...string) error

func (*Plumber) PipelineDeleteAll

func (p *Plumber) PipelineDeleteAll() error

func (*Plumber) Pipelines

func (p *Plumber) Pipelines() []string

Pipelines returns a sorted list of pipeline production strings

func (*Plumber) Pipes

func (p *Plumber) Pipes() []*Pipe

func (*Plumber) Plumb

func (p *Plumber) Plumb(production ...string) error

func (*Plumber) Via

func (p *Plumber) Via(pipe string, command []string)

func (*Plumber) Write

func (p *Plumber) Write(pipe string, value string)

type Reader

type Reader struct {
	C    chan string
	Done chan struct{}

	ID int64
	// contains filtered or unexported fields
}

func (*Reader) Close

func (r *Reader) Close()