direct

Documentation

Overview

Package direct contains the direct runner for running single-bundle pipelines in the current process. Useful for testing.

Index

• func Compile(edges []*graph.MultiEdge) (*exec.Plan, error)
• func Execute(ctx context.Context, p *beam.Pipeline) error
• type CoGBK
  • func (n *CoGBK) Down(ctx context.Context) error
  • func (n *CoGBK) FinishBundle(ctx context.Context) error
  • func (n *CoGBK) ID() exec.UnitID
  • func (n *CoGBK) ProcessElement(ctx context.Context, elm *exec.FullValue, _ ...exec.ReStream) error
  • func (n *CoGBK) StartBundle(ctx context.Context, id string, data exec.DataContext) error
  • func (n *CoGBK) String() string
  • func (n *CoGBK) Up(ctx context.Context) error
• type Impulse
  • func (n *Impulse) Down(ctx context.Context) error
  • func (n *Impulse) FinishBundle(ctx context.Context) error
  • func (n *Impulse) ID() exec.UnitID
  • func (n *Impulse) Process(ctx context.Context) error
  • func (n *Impulse) StartBundle(ctx context.Context, id string, data exec.DataContext) error
  • func (n *Impulse) String() string
  • func (n *Impulse) Up(ctx context.Context) error
• type Inject
  • func (n *Inject) Down(ctx context.Context) error
  • func (n *Inject) FinishBundle(ctx context.Context) error
  • func (n *Inject) ID() exec.UnitID
  • func (n *Inject) ProcessElement(ctx context.Context, elm *exec.FullValue, values ...exec.ReStream) error
  • func (n *Inject) StartBundle(ctx context.Context, id string, data exec.DataContext) error
  • func (n *Inject) String() string
  • func (n *Inject) Up(ctx context.Context) error

Functions

func Compile

func Compile(edges []*graph.MultiEdge) (*exec.Plan, error)

Compile translates a pipeline to a multi-bundle execution plan.

func Execute

func Execute(ctx context.Context, p *beam.Pipeline) error

Execute runs the pipeline in-process.

Types

type CoGBK

type CoGBK struct {
	UID  exec.UnitID
	Edge *graph.MultiEdge
	Out  exec.Node
	// contains filtered or unexported fields
}

CoGBK buffers all input and continues on FinishBundle. Use with small single-bundle data only.

func (*CoGBK) Down

func (n *CoGBK) Down(ctx context.Context) error

func (*CoGBK) FinishBundle

func (n *CoGBK) FinishBundle(ctx context.Context) error

func (*CoGBK) ID

func (n *CoGBK) ID() exec.UnitID

func (*CoGBK) ProcessElement

func (n *CoGBK) ProcessElement(ctx context.Context, elm *exec.FullValue, _ ...exec.ReStream) error

func (*CoGBK) StartBundle

func (n *CoGBK) StartBundle(ctx context.Context, id string, data exec.DataContext) error

func (*CoGBK) String

func (n *CoGBK) String() string

func (*CoGBK) Up

func (n *CoGBK) Up(ctx context.Context) error

type Impulse

type Impulse struct {
	UID   exec.UnitID
	Value []byte
	Out   exec.Node
}

Impulse emits its single element in one invocation.

func (*Impulse) Down

func (n *Impulse) Down(ctx context.Context) error

func (*Impulse) FinishBundle

func (n *Impulse) FinishBundle(ctx context.Context) error

func (*Impulse) ID

func (n *Impulse) ID() exec.UnitID

func (*Impulse) Process

func (n *Impulse) Process(ctx context.Context) error

func (*Impulse) StartBundle

func (n *Impulse) StartBundle(ctx context.Context, id string, data exec.DataContext) error

func (*Impulse) String

func (n *Impulse) String() string

func (*Impulse) Up

func (n *Impulse) Up(ctx context.Context) error

type Inject

type Inject struct {
	UID exec.UnitID
	N   int
	Out exec.Node
}

Inject injects the predecessor index into each FullValue, effectively converting KV<X,Y> into KV<X,KV<int,Y>>. Used to prime CoGBK.

func (*Inject) Down

func (n *Inject) Down(ctx context.Context) error

func (*Inject) FinishBundle

func (n *Inject) FinishBundle(ctx context.Context) error

func (*Inject) ID

func (n *Inject) ID() exec.UnitID

func (*Inject) ProcessElement

func (n *Inject) ProcessElement(ctx context.Context, elm *exec.FullValue, values ...exec.ReStream) error

func (*Inject) StartBundle

func (n *Inject) StartBundle(ctx context.Context, id string, data exec.DataContext) error

func (*Inject) String

func (n *Inject) String() string

func (*Inject) Up

func (n *Inject) Up(ctx context.Context) error