streaming

Documentation

Index

• Variables
• func New(source flow.Source) flow.Flow
• type TopNOption
  • func OrderBy(sort TopNSort) TopNOption
  • func WithSortKeyExtractor(sortKeyExtractor func(flow.StreamRecord) int64) TopNOption
• type TopNSort
• type TriggerResult
• type TumblingTimeWindows
  • func NewTumblingTimeWindows(size time.Duration) *TumblingTimeWindows
  • func (s *TumblingTimeWindows) AssignWindows(timestamp int64) ([]flow.Window, error)
  • func (s *TumblingTimeWindows) Exec(downstream flow.Inlet)
  • func (s *TumblingTimeWindows) In() chan<- flow.StreamRecord
  • func (s *TumblingTimeWindows) Out() <-chan flow.StreamRecord
  • func (s *TumblingTimeWindows) Setup(ctx context.Context) (err error)
  • func (s *TumblingTimeWindows) Teardown(ctx context.Context) error
• type Tuple2
  • func (t *Tuple2) Equal(other *Tuple2) bool

Variables

var (
	DefaultCacheSize = 2
)

Functions

func New

func New(source flow.Source) flow.Flow

Types

type TopNOption

type TopNOption func(aggregator *topNAggregator)

func OrderBy

func OrderBy(sort TopNSort) TopNOption

func WithSortKeyExtractor

func WithSortKeyExtractor(sortKeyExtractor func(flow.StreamRecord) int64) TopNOption

type TopNSort

type TopNSort uint8

const (
	DESC TopNSort = iota
	ASC
)

type TriggerResult

type TriggerResult bool

const (
	FIRE     TriggerResult = true
	CONTINUE               = false
)

type TumblingTimeWindows

type TumblingTimeWindows struct {
	// internal state of the sliding time window
	flow.ComponentState
	// contains filtered or unexported fields
}

func NewTumblingTimeWindows

func NewTumblingTimeWindows(size time.Duration) *TumblingTimeWindows

func (*TumblingTimeWindows) AssignWindows

func (s *TumblingTimeWindows) AssignWindows(timestamp int64) ([]flow.Window, error)

AssignWindows assigns windows according to the given timestamp

func (*TumblingTimeWindows) Exec

func (s *TumblingTimeWindows) Exec(downstream flow.Inlet)

func (*TumblingTimeWindows) In

func (s *TumblingTimeWindows) In() chan<- flow.StreamRecord

func (*TumblingTimeWindows) Out

func (s *TumblingTimeWindows) Out() <-chan flow.StreamRecord

func (*TumblingTimeWindows) Setup

func (s *TumblingTimeWindows) Setup(ctx context.Context) (err error)

func (*TumblingTimeWindows) Teardown

func (s *TumblingTimeWindows) Teardown(ctx context.Context) error

type Tuple2

type Tuple2 struct {
	V1 interface{} `json:"v1"`
	V2 interface{} `json:"v2"`
}

func (*Tuple2) Equal

func (t *Tuple2) Equal(other *Tuple2) bool