sliding

Documentation

Overview

Package sliding implements Sliding windows. Sliding windows are defined by a static window size e.g. minutely windows or hourly windows and a fixed "slide". This is the duration by which the boundaries of the windows move once every <slide> duration. Package sliding also maintains the state of active windows. Watermark is used to trigger the expiration of windows.

Index

• func NewSlidingWindow(startTime time.Time, endTime time.Time) window.TimedWindow
• func NewWindower(length time.Duration, slide time.Duration, vertexInstance *dfv1.VertexInstance) window.TimedWindower
• type Windower
  • func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindowRequest
  • func (w *Windower) CloseWindows(time time.Time) []*window.TimedWindowRequest
  • func (w *Windower) DeleteClosedWindow(window window.TimedWindow)
  • func (w *Windower) InsertWindow(tw window.TimedWindow)
  • func (w *Windower) NextWindowToBeClosed() window.TimedWindow
  • func (w *Windower) OldestWindowEndTime() time.Time
  • func (w *Windower) Strategy() window.Strategy
  • func (*Windower) Type() window.Type

Functions

func NewSlidingWindow

func NewSlidingWindow(startTime time.Time, endTime time.Time) window.TimedWindow

func NewWindower

func NewWindower(length time.Duration, slide time.Duration, vertexInstance *dfv1.VertexInstance) window.TimedWindower

Types

type Windower

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

Windower is an implementation of TimedWindower of sliding window, windower is responsible for assigning windows to the incoming messages and closing the windows that are past the watermark.

func (*Windower) AssignWindows

func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindowRequest

AssignWindows assigns the event to the window based on give window configuration. For sliding window, the message can be assigned to multiple windows. The operation can be either OPEN or APPEND, depending on whether the window is already present or not.

func (*Windower) CloseWindows

func (w *Windower) CloseWindows(time time.Time) []*window.TimedWindowRequest

CloseWindows closes the windows that are past the watermark.

func (*Windower) DeleteClosedWindow

func (w *Windower) DeleteClosedWindow(window window.TimedWindow)

DeleteClosedWindow deletes the window from the closed windows list

func (*Windower) InsertWindow

func (w *Windower) InsertWindow(tw window.TimedWindow)

InsertWindow inserts a window to the list of active windows.

func (*Windower) NextWindowToBeClosed

func (w *Windower) NextWindowToBeClosed() window.TimedWindow

NextWindowToBeClosed returns the next window yet to be closed.

func (*Windower) OldestWindowEndTime

func (w *Windower) OldestWindowEndTime() time.Time

OldestWindowEndTime returns the end time of the oldest window among both active and closed windows. If there are no windows, it returns -1.

func (*Windower) Strategy

func (w *Windower) Strategy() window.Strategy

func (*Windower) Type

func (*Windower) Type() window.Type

Type implements window.TimedWindower.