blob: 6c4b98f013abc0d5da1d83fd5a1976b9bd871959 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
package core
import "fmt"
type Window interface {
// MaxTimestamp is the maximum time the Window is Open
MaxTimestamp() int64
// ToString convert Window to string format
ToString() string
}
type TimeWindow struct {
Start int64
End int64
}
func (t TimeWindow) MaxTimestamp() int64 {
// MaxTimestamp is 1ms before the End
return t.End - 1
}
func (t TimeWindow) ToString() string {
return fmt.Sprintf("timewindow(%d,%d)", t.Start, t.End)
}
type WindowAssigner[IN any] interface {
AssignWindows(element IN, timestamp int64) []Window
}
type TumblingWindows[IN any] struct {
Size int64
}
func NewTumblingWindows[IN any](size int64) *TumblingWindows[IN] {
return &TumblingWindows[IN]{
Size: size,
}
}
func (w *TumblingWindows[IN]) AssignWindows(element IN, timestamp int64) []Window {
// module to get start end
// if timestamp = 1200, size 1000
// so start will 1200 - 200 = 1000, end will 1000 + 1000 = 2000
start := timestamp - (timestamp % w.Size)
end := start + w.Size
return []Window{
TimeWindow{
Start: start,
End: end,
},
}
}
|
