window

Dec 20, 2025

Changes in this version

type SessionWindow

+ func (sw *SessionWindow) GetStats() map[string]int64

+ func (sw *SessionWindow) ResetStats()

type Window

+ GetStats func() map[string]int64

Nov 15, 2025

Changes in this version

+ var EnableDebug = false

type CountingWindow

+ func (cw *CountingWindow) GetStats() map[string]int64

+ func (cw *CountingWindow) Stop()

+ type Watermark struct

+ func NewWatermark(maxOutOfOrderness time.Duration, updateInterval time.Duration, ...) *Watermark

+ func (wm *Watermark) GetCurrentWatermark() time.Time

+ func (wm *Watermark) IsEventTimeLate(eventTime time.Time) bool

+ func (wm *Watermark) Stop()

+ func (wm *Watermark) UpdateEventTime(eventTime time.Time)

+ func (wm *Watermark) WatermarkChan() <-chan time.Time

type Window

+ Stop func()

Jul 27, 2025

Changes in this version

+ const TypeCounting

+ const TypeSession

+ const TypeSliding

+ const TypeTumbling

+ func GetTimestamp(data interface{}, tsProp string, timeUnit time.Duration) time.Time

+ type CountingWindow struct

+ func NewCountingWindow(config types.WindowConfig) (*CountingWindow, error)

+ func (cw *CountingWindow) Add(data interface{})

+ func (cw *CountingWindow) OutputChan() <-chan []types.Row

+ func (cw *CountingWindow) Reset()

+ func (cw *CountingWindow) SetCallback(callback func([]types.Row))

+ func (cw *CountingWindow) Start()

+ func (cw *CountingWindow) Trigger()

+ type SessionWindow struct

+ func NewSessionWindow(config types.WindowConfig) (*SessionWindow, error)

+ func (sw *SessionWindow) Add(data interface{})

+ func (sw *SessionWindow) OutputChan() <-chan []types.Row

+ func (sw *SessionWindow) Reset()

+ func (sw *SessionWindow) SetCallback(callback func([]types.Row))

+ func (sw *SessionWindow) Start()

+ func (sw *SessionWindow) Stop()

+ func (sw *SessionWindow) Trigger()

+ type SlidingWindow struct

+ func NewSlidingWindow(config types.WindowConfig) (*SlidingWindow, error)

+ func (sw *SlidingWindow) Add(data interface{})

+ func (sw *SlidingWindow) GetStats() map[string]int64

+ func (sw *SlidingWindow) NextSlot() *types.TimeSlot

+ func (sw *SlidingWindow) OutputChan() <-chan []types.Row

+ func (sw *SlidingWindow) Reset()

+ func (sw *SlidingWindow) ResetStats()

+ func (sw *SlidingWindow) SetCallback(callback func([]types.Row))

+ func (sw *SlidingWindow) Start()

+ func (sw *SlidingWindow) Stop()

+ func (sw *SlidingWindow) Trigger()

+ type TimedData struct

+ Data interface{}

+ Timestamp time.Time

+ type TumblingWindow struct

+ func NewTumblingWindow(config types.WindowConfig) (*TumblingWindow, error)

+ func (sw *TumblingWindow) NextSlot() *types.TimeSlot

+ func (tw *TumblingWindow) Add(data interface{})

+ func (tw *TumblingWindow) GetStats() map[string]int64

+ func (tw *TumblingWindow) OutputChan() <-chan []types.Row

+ func (tw *TumblingWindow) Reset()

+ func (tw *TumblingWindow) ResetStats()

+ func (tw *TumblingWindow) SetCallback(callback func([]types.Row))

+ func (tw *TumblingWindow) Start()

+ func (tw *TumblingWindow) Stop()

+ func (tw *TumblingWindow) Trigger()

+ type Window interface

+ Add func(item interface{})

+ OutputChan func() <-chan []types.Row

+ Reset func()

+ SetCallback func(callback func([]types.Row))

+ Start func()

+ Trigger func()

+ func CreateWindow(config types.WindowConfig) (Window, error)