sliding_window

Documentation

Index

• Constants
• Variables
• type SequenceGenerator
  • func NewSequenceGenerator() *SequenceGenerator
  • func (generator *SequenceGenerator) Next() int64
• type Watermark
  • func (w Watermark) Healthy() bool
• type Window
  • func NewStaticSlidingWindow(windowSize int, name string) Window
• type WindowItem

Constants

const FirstSequenceNumber = 1

const ProcessDelayWarningThreshold = 30

Variables

var HealthyThreshold float64 = 60

Types

type SequenceGenerator

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

func NewSequenceGenerator

func NewSequenceGenerator() *SequenceGenerator

func (*SequenceGenerator) Next

func (generator *SequenceGenerator) Next() int64

type Watermark

type Watermark struct {
	ProcessTime time.Time
	EventTime   time.Time
}

(Output) Watermark is defined as the minimum process time of input(which may be blocked on enqueue) and active items in window.

func (Watermark) Healthy

func (w Watermark) Healthy() bool

type Window

type Window interface {
	AddWindowItem(item WindowItem)
	AckWindowItem(sequence int64)
	Size() int
	WaitingQueueLen() int
	Close()
	Watermark() Watermark
}

func NewStaticSlidingWindow

func NewStaticSlidingWindow(windowSize int, name string) Window

type WindowItem

type WindowItem interface {
	SequenceNumber() int64
	BeforeWindowMoveForward()
	EventTime() time.Time
	ProcessTime() time.Time
}