pipeline

Documentation

Index

• Variables
• func GetBlock() *[]byte
• func PutBlock(b *[]byte)
• func SetupBlockPool(sz int)
• type BlockBufferReader
  • func (b *BlockBufferReader) Get() ([]byte, error)
  • func (b *BlockBufferReader) Init(buf *[]byte)
  • func (b *BlockBufferReader) Len() int
• type BlockBufferWriter
  • func (b *BlockBufferWriter) Close()
  • func (b *BlockBufferWriter) Init(buf *[]byte)
  • func (b *BlockBufferWriter) IsEmpty() bool
  • func (b *BlockBufferWriter) Put(itms ...[]byte) error
• type Filter
• type ItemReadWriter
  • func (rw *ItemReadWriter) InitReadWriter()
  • func (rw *ItemReadWriter) Kill()
• type ItemReader
  • func (r *ItemReader) CloseRead() error
  • func (r *ItemReader) FlushBlock()
  • func (r *ItemReader) InitReader()
  • func (r *ItemReader) Kill()
  • func (r *ItemReader) PeekBlock() ([]byte, error)
  • func (r *ItemReader) ReadItem() ([]byte, error)
  • func (r *ItemReader) SetSource(w Writer)
• type ItemWriter
  • func (w *ItemWriter) Channel() chan interface{}
  • func (w *ItemWriter) CloseWithError(err error)
  • func (w *ItemWriter) CloseWrite() error
  • func (w *ItemWriter) HasShutdown() error
  • func (w *ItemWriter) InitWriter()
  • func (w *ItemWriter) Kill()
  • func (w *ItemWriter) ResizeBlockBuffer(itmLen int)
  • func (w *ItemWriter) SetNumBuffers(n int)
  • func (w *ItemWriter) Shutdown(err error)
  • func (w *ItemWriter) WriteItem(itm ...[]byte) error
• type Pipeline
  • func (p *Pipeline) AddFilter(name string, f Filter) error
  • func (p *Pipeline) AddSink(name string, s Sink) error
  • func (p *Pipeline) AddSource(name string, s Source) error
  • func (p *Pipeline) Execute() error
  • func (p *Pipeline) Finalize()
• type Runnable
• type Sink
• type Source
• type Writer

Variables

var (
	ErrNoBlockSpace = errors.New("Not enough space in buffer")
	ErrNoMoreItem   = errors.New("No more item to be read from buffer")
)

var (
	ErrSinkExist      = errors.New("Sink already exist")
	ErrSupervisorKill = errors.New("Supervisor requested exit")
)

Functions

func GetBlock

func GetBlock() *[]byte

func PutBlock

func PutBlock(b *[]byte)

func SetupBlockPool

func SetupBlockPool(sz int)

Types

type BlockBufferReader

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

func (*BlockBufferReader) Get

func (b *BlockBufferReader) Get() ([]byte, error)

func (*BlockBufferReader) Init

func (b *BlockBufferReader) Init(buf *[]byte)

func (*BlockBufferReader) Len

func (b *BlockBufferReader) Len() int

type BlockBufferWriter

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

func (*BlockBufferWriter) Close

func (b *BlockBufferWriter) Close()

func (*BlockBufferWriter) Init

func (b *BlockBufferWriter) Init(buf *[]byte)

func (*BlockBufferWriter) IsEmpty

func (b *BlockBufferWriter) IsEmpty() bool

func (*BlockBufferWriter) Put

func (b *BlockBufferWriter) Put(itms ...[]byte) error

type Filter

type Filter interface {
	Writer
	SetSource(Writer)
	Routine() error
}

type ItemReadWriter

type ItemReadWriter struct {
	ItemReader
	ItemWriter
}

func (*ItemReadWriter) InitReadWriter

func (rw *ItemReadWriter) InitReadWriter()

func (*ItemReadWriter) Kill

func (rw *ItemReadWriter) Kill()

type ItemReader

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

func (*ItemReader) CloseRead

func (r *ItemReader) CloseRead() error

func (*ItemReader) FlushBlock

func (r *ItemReader) FlushBlock()

func (*ItemReader) InitReader

func (r *ItemReader) InitReader()

func (*ItemReader) Kill

func (r *ItemReader) Kill()

func (*ItemReader) PeekBlock

func (r *ItemReader) PeekBlock() ([]byte, error)

func (*ItemReader) ReadItem

func (r *ItemReader) ReadItem() ([]byte, error)

func (*ItemReader) SetSource

func (r *ItemReader) SetSource(w Writer)

type ItemWriter

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

func (*ItemWriter) Channel

func (w *ItemWriter) Channel() chan interface{}

func (*ItemWriter) CloseWithError

func (w *ItemWriter) CloseWithError(err error)

func (*ItemWriter) CloseWrite

func (w *ItemWriter) CloseWrite() error

func (*ItemWriter) HasShutdown

func (w *ItemWriter) HasShutdown() error

func (*ItemWriter) InitWriter

func (w *ItemWriter) InitWriter()

func (*ItemWriter) Kill

func (w *ItemWriter) Kill()

func (*ItemWriter) ResizeBlockBuffer

func (w *ItemWriter) ResizeBlockBuffer(itmLen int)

func (*ItemWriter) SetNumBuffers

func (w *ItemWriter) SetNumBuffers(n int)

func (*ItemWriter) Shutdown

func (w *ItemWriter) Shutdown(err error)

func (*ItemWriter) WriteItem

func (w *ItemWriter) WriteItem(itm ...[]byte) error

type Pipeline

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

func (*Pipeline) AddFilter

func (p *Pipeline) AddFilter(name string, f Filter) error

func (*Pipeline) AddSink

func (p *Pipeline) AddSink(name string, s Sink) error

func (*Pipeline) AddSource

func (p *Pipeline) AddSource(name string, s Source) error

func (*Pipeline) Execute

func (p *Pipeline) Execute() error

func (*Pipeline) Finalize

func (p *Pipeline) Finalize()

type Runnable

type Runnable interface {
	Routine() error
	Kill()
}

type Sink

type Sink interface {
	SetSource(Writer)
	Routine() error
}

type Source

type Source interface {
	Writer
	Shutdown(err error)
	Routine() error
}

type Writer

type Writer interface {
	Channel() chan interface{}
}