feat: add on timer & watermarkmain

3 files changed, 175 insertions, 0 deletions

diff --git a/core/operator/filter.go b/core/operator/filter.go
new file mode 100644
index 0000000..183148b
--- /dev/null
+++ b/core/operator/filter.go
@@ -0,0 +1,36 @@
+package operator
+
+import (
+	"context"
+	"goflink/core"
+)
+
+type FilterFunc[IN any] func(in IN) bool
+
+type FilterOperator[IN any] struct {
+	userFunc FilterFunc[IN]
+}
+
+func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] {
+	return &FilterOperator[IN]{userFunc: userFunc}
+}
+
+func (f *FilterOperator[IN]) Open(ctx context.Context) error {
+	return nil
+}
+
+func (f *FilterOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
+	if record.Type != core.RecordData {
+		return out.Emit(ctx, record)
+	}
+
+	if f.userFunc(record.Value) {
+		return out.Emit(ctx, record)
+	}
+
+	return nil
+}
+
+func (f *FilterOperator[IN]) Close(ctx context.Context) error {
+	return nil
+}
diff --git a/core/operator/keyed_process.go b/core/operator/keyed_process.go
new file mode 100644
index 0000000..b507d87
--- /dev/null
+++ b/core/operator/keyed_process.go
@@ -0,0 +1,81 @@
+package operator
+
+import (
+	"context"
+	"fmt"
+	"goflink/core"
+)
+
+type KeyedProcesser[IN, OUT any] interface {
+	Open(ctx context.Context) error
+	ProcessElement(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error
+	Close(ctx context.Context) error
+	OnTimer(ctx context.Context, timestamp int64, out Emitter[OUT]) error
+}
+
+type KeyProcessOperator[IN, OUT any] struct {
+	userFunc    KeyedProcesser[IN, OUT]
+	timeService core.InternalTimerService 
+	backend     core.StateBackend
+}
+
+func NewKeyProcessOperator[IN, OUT any](userFunc KeyedProcesser[IN, OUT]) *KeyProcessOperator[IN, OUT] {
+	return &KeyProcessOperator[IN, OUT]{
+		userFunc:    userFunc,
+		timeService: core.NewTimerService(),
+	}
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Open(ctx context.Context) error {
+	// init backend
+	be, ok := core.ExtractStateBackend(ctx)
+	if !ok {
+		return fmt.Errorf("state backend not found")
+	}
+
+	k.backend = be
+	// inject the timer to context for user use
+	ctx = core.InjectTimerService(ctx, k.timeService)
+	return k.userFunc.Open(ctx)
+
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+
+	switch record.Type {
+	case core.RecordData:
+		if err := k.backend.SetCurrentKey(record.Key); err != nil {
+			return err
+		}
+
+		return k.userFunc.ProcessElement(ctx, record, out)
+	case core.RecordWatermark:
+		// advance the timer
+		triggerTimerQueue := k.timeService.AdvanceWatermark(record.Timestamp)
+
+		for _, t := range triggerTimerQueue {
+
+			if err := k.backend.SetCurrentKey(t.Key); err != nil {
+				return err
+			}
+			if err := k.userFunc.OnTimer(ctx, t.Timestamp, out); err != nil {
+				return err
+			}
+		}
+		return out.Emit(ctx, core.StreamRecord[OUT]{
+			Type:      core.RecordWatermark,
+			Timestamp: record.Timestamp,
+		})
+	default:
+		return out.Emit(ctx, core.StreamRecord[OUT]{
+			Timestamp: record.Timestamp,
+			BarrierID: record.BarrierID,
+			Key:       record.Key,
+		})
+	}
+
+}
+
+func (k *KeyProcessOperator[IN, OUT]) Close(ctx context.Context) error {
+	return k.userFunc.Close(ctx)
+}
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
+}