aboutsummaryrefslogtreecommitdiff
path: root/core/window.go
diff options
context:
space:
mode:
Diffstat (limited to 'core/window.go')
-rw-r--r--core/window.go54
1 files changed, 54 insertions, 0 deletions
diff --git a/core/window.go b/core/window.go
new file mode 100644
index 0000000..6c4b98f
--- /dev/null
+++ b/core/window.go
@@ -0,0 +1,54 @@
1package core
2
3import "fmt"
4
5type Window interface {
6 // MaxTimestamp is the maximum time the Window is Open
7 MaxTimestamp() int64
8 // ToString convert Window to string format
9 ToString() string
10}
11
12type TimeWindow struct {
13 Start int64
14 End int64
15}
16
17func (t TimeWindow) MaxTimestamp() int64 {
18 // MaxTimestamp is 1ms before the End
19 return t.End - 1
20}
21
22func (t TimeWindow) ToString() string {
23 return fmt.Sprintf("timewindow(%d,%d)", t.Start, t.End)
24}
25
26type WindowAssigner[IN any] interface {
27 AssignWindows(element IN, timestamp int64) []Window
28}
29
30type TumblingWindows[IN any] struct {
31 Size int64
32}
33
34func NewTumblingWindows[IN any](size int64) *TumblingWindows[IN] {
35 return &TumblingWindows[IN]{
36 Size: size,
37 }
38}
39
40func (w *TumblingWindows[IN]) AssignWindows(element IN, timestamp int64) []Window {
41 // module to get start end
42 // if timestamp = 1200, size 1000
43 // so start will 1200 - 200 = 1000, end will 1000 + 1000 = 2000
44 start := timestamp - (timestamp % w.Size)
45
46 end := start + w.Size
47
48 return []Window{
49 TimeWindow{
50 Start: start,
51 End: end,
52 },
53 }
54}