Verified feat: init base draft support window

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 @@
+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,
+                },
+        }
+}