transformerpipeline

Documentation

Index

• Variables
• type TransformerPipeline
  • func New(ctx context.Context, transformClients []plugin.Plugin_TransformClient) (*TransformerPipeline, context.Context, error)
  • func (lp *TransformerPipeline) Close()
  • func (lp *TransformerPipeline) OnOutput(fn func([]byte) error) error
  • func (lp *TransformerPipeline) RunBlocking() error
  • func (lp *TransformerPipeline) Send(data []byte) error

Variables

var (
	ErrPipelineClosed = errors.New("pipeline is closed")
)

Types

type TransformerPipeline

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

TransformerPipeline runs a pipeline of transform clients.

Ideally we'd just call the result of each transform to the next one, but transformations are not synchronous calls, so orchestration is needed. That's what this does: it hides the orchestration of the transform clients.

Use it like this:

- Construct a new TransformerPipeline with `New`. Give it a context and a slice of transform clients. - Register a callback for transformed records with `OnOutput`. - Start all transformers with `RunBlocking`. - Send records to the pipeline with `Send`. - When done, close the pipeline with `Close`. Otherwise, `RunBlocking` won't finish.

func New

func New(ctx context.Context, transformClients []plugin.Plugin_TransformClient) (*TransformerPipeline, context.Context, error)

func (*TransformerPipeline) Close

func (lp *TransformerPipeline) Close()

func (*TransformerPipeline) OnOutput

func (lp *TransformerPipeline) OnOutput(fn func([]byte) error) error

func (*TransformerPipeline) RunBlocking

func (lp *TransformerPipeline) RunBlocking() error

func (*TransformerPipeline) Send

func (lp *TransformerPipeline) Send(data []byte) error