plugins

Documentation

Index

• type Destination
  • func (d *Destination) Configure(opts []aql.Option) error
  • func (d *Destination) Open(s engine.Stream, l engine.Logger, st engine.Stopper)
  • func (d *Destination) Ping() error
  • func (d *Destination) SetInputColumns(source string, columns []string) error
  • func (d *Destination) SetName(name string)
• type DestinationJSONRPC
  • func (t *DestinationJSONRPC) Close() error
  • func (t *DestinationJSONRPC) Dial() error
  • func (t *DestinationJSONRPC) EOS() ([]LogEntry, error)
  • func (t *DestinationJSONRPC) Send(rows []InputRow) ([]LogEntry, error)
  • func (t *DestinationJSONRPC) SetInputColumns(source string, columns []string) error
  • func (t *DestinationJSONRPC) SetOption(name string, value interface{}) error
  • func (t *DestinationJSONRPC) SetSources(names []string) error
• type DestinationPlugin
• type InputRow
• type LogEntry
• type OutputRow
• type Plugin
• type Source
  • func (so *Source) Configure(opts []aql.Option) error
  • func (so *Source) Open(s engine.Stream, l engine.Logger, st engine.Stopper)
  • func (so *Source) Ping() error
  • func (so *Source) SetName(name string)
• type SourceJSONRPC
  • func (t *SourceJSONRPC) Close() error
  • func (t *SourceJSONRPC) Dial() error
  • func (t *SourceJSONRPC) GetOutputColumns() (map[string][]string, error)
  • func (t *SourceJSONRPC) Receive() ([]OutputRow, []LogEntry, error)
  • func (t *SourceJSONRPC) SetDestinations(names []string) error
  • func (t *SourceJSONRPC) SetOption(name string, value interface{}) error
• type SourcePlugin
• type Transform
  • func (d *Transform) Configure(opts []aql.Option) error
  • func (d *Transform) Open(s engine.Stream, dest engine.Stream, l engine.Logger, st engine.Stopper)
  • func (d *Transform) Ping() error
  • func (d *Transform) Sequence(sourceSeq []string)
  • func (d *Transform) SetInputColumns(source string, columns []string) error
  • func (d *Transform) SetName(name string)
• type TransformJSONRPC
  • func (t *TransformJSONRPC) Close() error
  • func (t *TransformJSONRPC) Dial() error
  • func (t *TransformJSONRPC) EOS() ([]OutputRow, []LogEntry, error)
  • func (t *TransformJSONRPC) GetOutputColumns() (map[string][]string, error)
  • func (t *TransformJSONRPC) Send(rows []InputRow) ([]OutputRow, []LogEntry, error)
  • func (t *TransformJSONRPC) SetDestinations(names []string) error
  • func (t *TransformJSONRPC) SetInputColumns(source string, columns []string) error
  • func (t *TransformJSONRPC) SetOption(name string, value interface{}) error
  • func (t *TransformJSONRPC) SetSources(names []string) error
• type TransformPlugin

Types

type Destination

type Destination struct {
	Plugin DestinationPlugin
	// contains filtered or unexported fields
}

Destination is the default implementation of a DestinationPlugin plugin that also satisfies the engine.DestinationPlugin interface.

func (*Destination) Configure

func (d *Destination) Configure(opts []aql.Option) error

func (*Destination) Open

func (d *Destination) Open(s engine.Stream, l engine.Logger, st engine.Stopper)

func (*Destination) Ping

func (d *Destination) Ping() error

func (*Destination) SetInputColumns

func (d *Destination) SetInputColumns(source string, columns []string) error

func (*Destination) SetName

func (d *Destination) SetName(name string)

type DestinationJSONRPC

type DestinationJSONRPC struct {
	Path string
	Args []string
	// contains filtered or unexported fields
}

func (*DestinationJSONRPC) Close

func (t *DestinationJSONRPC) Close() error

func (*DestinationJSONRPC) Dial

func (t *DestinationJSONRPC) Dial() error

func (*DestinationJSONRPC) EOS

func (t *DestinationJSONRPC) EOS() ([]LogEntry, error)

func (*DestinationJSONRPC) Send

func (t *DestinationJSONRPC) Send(rows []InputRow) ([]LogEntry, error)

func (*DestinationJSONRPC) SetInputColumns

func (t *DestinationJSONRPC) SetInputColumns(source string, columns []string) error

func (*DestinationJSONRPC) SetOption

func (t *DestinationJSONRPC) SetOption(name string, value interface{}) error

func (*DestinationJSONRPC) SetSources

func (t *DestinationJSONRPC) SetSources(names []string) error

type DestinationPlugin

type DestinationPlugin interface {
	Plugin

	//SetSources sets the names of the input sources.
	SetSources(names []string) error

	//SetInputColumns sets the names of the input columns for the given Source.
	SetInputColumns(source string, columns []string) error

	//Send sends a batch of rows to the plugin, optionally returning output rows and/or
	//log entries.
	Send(row []InputRow) ([]LogEntry, error)

	//EOS signals the end of the stream and that the plugin should exit.
	EOS() ([]LogEntry, error)
}

DestinationPlugin is the interface for destinations.

type InputRow

type InputRow struct {
	Source string        `json:"source"`
	Data   []interface{} `json:"data"`
}

InputRow is a row sent from the executor to the plugin.

type LogEntry

type LogEntry struct {
	Level   string `json:"level"`
	Message string `json:"message"`
}

LogEntry is a log entry recorded by the plugin.

type OutputRow

type OutputRow struct {
	Destination string        `json:"destination"`
	Data        []interface{} `json:"data"`
}

OutputRow is a row sent from the plugin to the executor.

type Plugin

type Plugin interface {
	//Dial connects to the plugin using whatever RPC. It can hold resources open.
	//These should be released when Close() is called.
	Dial() error

	//SetOption sets the given option name/value pair.
	SetOption(name string, value interface{}) error

	//Close releases any resources associated with the plugin.
	Close() error
}

Plugin is the generic interface that all plugins must satisfy.

type Source

type Source struct {
	Plugin SourcePlugin
	// contains filtered or unexported fields
}

Source is the default implementation of a SourcePlugin plugin that also satisfies the engine.SourcePlugin interface.

func (*Source) Configure

func (so *Source) Configure(opts []aql.Option) error

func (*Source) Open

func (so *Source) Open(s engine.Stream, l engine.Logger, st engine.Stopper)

func (*Source) Ping

func (so *Source) Ping() error

func (*Source) SetName

func (so *Source) SetName(name string)

type SourceJSONRPC

type SourceJSONRPC struct {
	Path string
	Args []string
	// contains filtered or unexported fields
}

func (*SourceJSONRPC) Close

func (t *SourceJSONRPC) Close() error

func (*SourceJSONRPC) Dial

func (t *SourceJSONRPC) Dial() error

func (*SourceJSONRPC) GetOutputColumns

func (t *SourceJSONRPC) GetOutputColumns() (map[string][]string, error)

func (*SourceJSONRPC) Receive

func (t *SourceJSONRPC) Receive() ([]OutputRow, []LogEntry, error)

func (*SourceJSONRPC) SetDestinations

func (t *SourceJSONRPC) SetDestinations(names []string) error

func (*SourceJSONRPC) SetOption

func (t *SourceJSONRPC) SetOption(name string, value interface{}) error

type SourcePlugin

type SourcePlugin interface {
	Plugin

	//SetDestinations sets the names of the output destinations.
	SetDestinations(names []string) error

	//GetOutputColumns gets the name of the output columns for the given Destination.
	GetOutputColumns() (map[string][]string, error)

	//Receive optionally returns output rows and/or log entries. The boolean parameter
	//is used to indicate whether End of Stream has been reached.
	Receive() ([]OutputRow, []LogEntry, error)
}

SourcePlugin is the interface for sources.

type Transform

type Transform struct {
	Plugin TransformPlugin
	Alias  string
	// contains filtered or unexported fields
}

Transform is the default implementation of a Transform plugin that also satisfies the engine.Transform interface.

func (*Transform) Configure

func (d *Transform) Configure(opts []aql.Option) error

func (*Transform) Open

func (d *Transform) Open(s engine.Stream, dest engine.Stream, l engine.Logger, st engine.Stopper)

func (*Transform) Ping

func (d *Transform) Ping() error

func (*Transform) Sequence

func (d *Transform) Sequence(sourceSeq []string)

func (*Transform) SetInputColumns

func (d *Transform) SetInputColumns(source string, columns []string) error

func (*Transform) SetName

func (d *Transform) SetName(name string)

type TransformJSONRPC

type TransformJSONRPC struct {
	Path string
	Args []string
	// contains filtered or unexported fields
}

func (*TransformJSONRPC) Close

func (t *TransformJSONRPC) Close() error

func (*TransformJSONRPC) Dial

func (t *TransformJSONRPC) Dial() error

func (*TransformJSONRPC) EOS

func (t *TransformJSONRPC) EOS() ([]OutputRow, []LogEntry, error)

func (*TransformJSONRPC) GetOutputColumns

func (t *TransformJSONRPC) GetOutputColumns() (map[string][]string, error)

func (*TransformJSONRPC) Send

func (t *TransformJSONRPC) Send(rows []InputRow) ([]OutputRow, []LogEntry, error)

func (*TransformJSONRPC) SetDestinations

func (t *TransformJSONRPC) SetDestinations(names []string) error

func (*TransformJSONRPC) SetInputColumns

func (t *TransformJSONRPC) SetInputColumns(source string, columns []string) error

func (*TransformJSONRPC) SetOption

func (t *TransformJSONRPC) SetOption(name string, value interface{}) error

func (*TransformJSONRPC) SetSources

func (t *TransformJSONRPC) SetSources(names []string) error

type TransformPlugin

type TransformPlugin interface {
	Plugin

	//SetSources sets the names of the input sources.
	SetSources(names []string) error

	//SetDestinations sets the names of the output destinations.
	SetDestinations(names []string) error

	//SetInputColumns sets the names of the input columns for the given Source.
	SetInputColumns(source string, columns []string) error

	//GetOutputColumns gets the name of the output columns for the given Destination.
	GetOutputColumns() (map[string][]string, error)

	//Send sends a batch of rows to the plugin, optionally returning output rows and/or
	//log entries.
	Send(row []InputRow) ([]OutputRow, []LogEntry, error)

	//EOS signals the end of the stream and that the plugin should exit.
	EOS() ([]OutputRow, []LogEntry, error)
}

TransformPlugin is the interface for transforms.