pipeline

Documentation

Overview

Package pipeline provides simple implementations of Pipeline for use in pre-processing streamline.Stream output.

Index

• type Filter
  • func (f Filter) ProcessLine(line []byte) ([]byte, error)
• type Map
  • func (m Map) ProcessLine(line []byte) ([]byte, error)
• type MapErr
  • func (m MapErr) ProcessLine(line []byte) ([]byte, error)
• type MultiPipeline
  • func (mp MultiPipeline) ProcessLine(line []byte) ([]byte, error)
• type Pipeline
  • func MapIdx(mapper func(i int, line []byte) ([]byte, error)) Pipeline
  • func Sample(n int) Pipeline

Types

type Filter

type Filter func(line []byte) bool

Filter is a Pipeline that allows omission of individual lines from streamline.Stream by returning false on lines that should not be included (i.e. should be skipped).

func (Filter) ProcessLine

func (f Filter) ProcessLine(line []byte) ([]byte, error)

type Map

type Map func(line []byte) []byte

Map is a Pipeline that allows modifications of individual lines from streamline.Stream. Implementations can return a nil []byte to indicate a line is to be skipped.

func (Map) ProcessLine

func (m Map) ProcessLine(line []byte) ([]byte, error)

type MapErr

type MapErr func(line []byte) ([]byte, error)

MapErr is a Pipeline that allows modifications of individual lines from streamline.Stream with error handling. Implementations can return a nil []byte to indicate a line is to be skipped.

Errors interrupt line processing and are propagated to streamline.Stream.

func (MapErr) ProcessLine

func (m MapErr) ProcessLine(line []byte) ([]byte, error)

type MultiPipeline

type MultiPipeline []Pipeline

MultiPipeline is a Pipeline that applies all its Pipelines in serial.

func (MultiPipeline) ProcessLine

func (mp MultiPipeline) ProcessLine(line []byte) ([]byte, error)

ProcessLine will provide the line to all active pipelines in the MultiPipeline in serial, passing the result of each pipeline to the next. If any pipeline indicates a line should be skipped by returning a nil line, then ProcessLine returns immediately.

type Pipeline

type Pipeline interface {
	// ProcessLine returns a modified, unmodified, or omitted line. To omit a line, return
	// a nil []byte - an empty []byte will cause an empty line to be retained.
	//
	// Implementations must not retain line.
	ProcessLine(line []byte) ([]byte, error)
}

Pipeline implementations are used to transform the data provided to a streamline.Stream. For example, they are useful for mapping and pruning data. To configure a Stream to use a Pipeline, use (*Stream).WithPipeline(...).

Note that generally a Pipeline should not be used to implement handling of data - use (*Stream).Stream(...) and (*Stream).StreamBytes(...) instead.

func MapIdx

func MapIdx(mapper func(i int, line []byte) ([]byte, error)) Pipeline

MapIdx is a Pipeline that allows modifications of individual lines from streamline.Stream based on the index of each line (i.e. how many lines the Pipeline has processed). The first line to be processed has an index of 0.

func Sample

func Sample(n int) Pipeline

Sample creates a Sampler pipeline that only includes every nth line from streamline.Stream. If N is 0, all lines are skipped; if N is 1, all lines are retained. Negative values will result in an error.