From 91b1c642601efbbcc5af9931e087f68b2c72fc3d Mon Sep 17 00:00:00 2001
From: SangTran-127 <tranquangsang12.7@gmail.com>
Date: Thu, 13 Aug 2026 23:42:53 +0700
Subject: feat: add on timer & watermark

---
 core/operator/watermark.go | 58 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 core/operator/watermark.go

(limited to 'core/operator/watermark.go')

diff --git a/core/operator/watermark.go b/core/operator/watermark.go
new file mode 100644
index 0000000..75390a8
--- /dev/null
+++ b/core/operator/watermark.go
@@ -0,0 +1,58 @@
+package operator
+
+import (
+	"context"
+	"goflink/core"
+)
+
+// TimestampAssigner is responsible for event time defined by user
+type TimestampAssigner[IN any] func(in IN) int64
+
+type WaterGeneratorOperator[IN any] struct {
+	assigner             TimestampAssigner[IN]
+	maxOutOfOrder        int64
+	currentMaxTimestamp  int64
+	lastEmittedWatermark int64
+}
+
+func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64, assigner TimestampAssigner[IN]) *WaterGeneratorOperator[IN] {
+	return &WaterGeneratorOperator[IN]{
+		maxOutOfOrder:        maxOutOfOrder,
+		assigner:             assigner,
+		lastEmittedWatermark: -1,
+	}
+}
+
+func (w *WaterGeneratorOperator[IN]) Open(ctx context.Context) error {
+	return nil
+}
+
+func (w *WaterGeneratorOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
+	if record.Type == core.RecordData {
+
+		// query the time
+		eventTime := w.assigner(record.Value)
+		record.Timestamp = eventTime
+
+		// update the max timestamp
+		if w.currentMaxTimestamp < eventTime {
+			w.currentMaxTimestamp = eventTime
+		}
+
+		// calc the watermark
+
+		watermark := w.currentMaxTimestamp - w.maxOutOfOrder
+
+		// check if current watermark later than previous watermark
+		if watermark > w.lastEmittedWatermark {
+			w.lastEmittedWatermark = watermark
+			return out.Emit(ctx, core.NewWatermarkRecord[IN](watermark))
+		}
+		return nil
+	}
+	return out.Emit(ctx, record)
+}
+
+func (w *WaterGeneratorOperator[IN]) Close(ctx context.Context) error {
+	return nil
+}
-- 
cgit v1.2.3