engine

Documentation

Overview

Package engine implements the actual query processor. It runs exacly one query generated using the compiler package.

Index

• Constants
• type Engine
  • func NewEngine(dataReader io.Reader, dataWriter io.Writer, planReader io.Reader, ...) *Engine
  • func (e *Engine) AggregateFilterWorker()
  • func (e *Engine) AggregateWorker()
  • func (e *Engine) EgressWorker()
  • func (e *Engine) IngressFilterWorker()
  • func (e *Engine) IngressWorker()
  • func (e *Engine) LiveDistanceWindowWorker()
  • func (e *Engine) LiveTimeWindowWorker()
  • func (e *Engine) ProjectFilterWorker()
  • func (e *Engine) ProjectWorker()
  • func (e *Engine) ReplayDistanceWindowWorker()
  • func (e *Engine) ReplayTimeWindowWorker()
  • func (e *Engine) Run()
  • func (e *Engine) SessionWindowWorker()
  • func (e *Engine) WindowWorker()
• type Window
• type WindowGroup
  • func CreateWindowGroup(groupFieldNames []string) (wg WindowGroup)
  • func (wg *WindowGroup) AllGroupKeys() (keys []string)
  • func (wg *WindowGroup) Append(ingressRow *data.IngressRow)
  • func (wg *WindowGroup) Close(groupKey string) (window Window, ok bool)
  • func (wg *WindowGroup) GroupKey(ingressRow *data.IngressRow) (key string)
  • func (wg *WindowGroup) IsOpen(groupKey string) (ok bool)

Constants

const (
	CsvComment          = '#'
	ChannelCapacity int = 1000
)

Types

type Engine

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

func NewEngine

func NewEngine(
	dataReader io.Reader,
	dataWriter io.Writer,
	planReader io.Reader,
	exitAfterSeconds int,
) *Engine

func (*Engine) AggregateFilterWorker

func (e *Engine) AggregateFilterWorker()

func (*Engine) AggregateWorker

func (e *Engine) AggregateWorker()

func (*Engine) EgressWorker

func (e *Engine) EgressWorker()

func (*Engine) IngressFilterWorker

func (e *Engine) IngressFilterWorker()

func (*Engine) IngressWorker

func (e *Engine) IngressWorker()

func (*Engine) LiveDistanceWindowWorker

func (e *Engine) LiveDistanceWindowWorker()

func (*Engine) LiveTimeWindowWorker

func (e *Engine) LiveTimeWindowWorker()

func (*Engine) ProjectFilterWorker

func (e *Engine) ProjectFilterWorker()

func (*Engine) ProjectWorker

func (e *Engine) ProjectWorker()

func (*Engine) ReplayDistanceWindowWorker

func (e *Engine) ReplayDistanceWindowWorker()

func (*Engine) ReplayTimeWindowWorker

func (e *Engine) ReplayTimeWindowWorker()

If we have historic data, we process it as fast as possible.

func (*Engine) Run

func (e *Engine) Run()

func (*Engine) SessionWindowWorker

func (e *Engine) SessionWindowWorker()

func (*Engine) WindowWorker

func (e *Engine) WindowWorker()

type Window

type Window []*data.IngressRow

type WindowGroup

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

func CreateWindowGroup

func CreateWindowGroup(groupFieldNames []string) (wg WindowGroup)

func (*WindowGroup) AllGroupKeys

func (wg *WindowGroup) AllGroupKeys() (keys []string)

func (*WindowGroup) Append

func (wg *WindowGroup) Append(ingressRow *data.IngressRow)

func (*WindowGroup) Close

func (wg *WindowGroup) Close(groupKey string) (window Window, ok bool)

func (*WindowGroup) GroupKey

func (wg *WindowGroup) GroupKey(ingressRow *data.IngressRow) (key string)

func (*WindowGroup) IsOpen

func (wg *WindowGroup) IsOpen(groupKey string) (ok bool)