8 files changed, 464 insertions, 0 deletions

diff --git a/core/operator/filter.go b/core/operator/filter.go
new file mode 100644
index 0000000..80bbb77
--- /dev/null
+++ b/core/operator/filter.go
@@ -0,0 +1,36 @@
+package operator
+
+import (
+        "context"
+        "flux/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/keyby.go b/core/operator/keyby.go
new file mode 100644
index 0000000..56ef34e
--- /dev/null
+++ b/core/operator/keyby.go
@@ -0,0 +1,32 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+)
+
+type KeySelector[IN any] func(in IN) string
+type KeyByOperator[IN any] struct {
+        selector KeySelector[IN]
+}
+
+func NewKeyByOperator[IN any](selector KeySelector[IN]) *KeyByOperator[IN] {
+        return &KeyByOperator[IN]{selector: selector}
+}
+
+func (k *KeyByOperator[IN]) Open(ctx context.Context) error {
+        return nil
+}
+
+func (k *KeyByOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error {
+        if record.Type != core.RecordData {
+                return out.Emit(ctx, record)
+        }
+
+        record.Key = k.selector(record.Value)
+        return out.Emit(ctx, record)
+}
+
+func (k *KeyByOperator[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..1d3c86c
--- /dev/null
+++ b/core/operator/keyed_process.go
@@ -0,0 +1,81 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+        "fmt"
+)
+
+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/map.go b/core/operator/map.go
new file mode 100644
index 0000000..2165c25
--- /dev/null
+++ b/core/operator/map.go
@@ -0,0 +1,53 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+        "fmt"
+)
+
+// MapFunc is business logic for user implement
+type MapFunc[IN, OUT any] func(in IN) (OUT, error)
+
+type MapOperator[IN, OUT any] struct {
+        userFunc MapFunc[IN, OUT]
+}
+
+func NewMapOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
+        return &MapOperator[IN, OUT]{
+                userFunc: fn,
+        }
+}
+
+func (fo *MapOperator[IN, OUT]) Open(ctx context.Context) error {
+        return nil
+}
+
+func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        if record.Type != core.RecordData {
+                //       watermark and barrier
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      record.Type,
+                        Key:       record.Key,
+                        Timestamp: record.Timestamp,
+                        BarrierID: record.BarrierID,
+                })
+        }
+
+        outValue, err := fo.userFunc(record.Value)
+
+        if err != nil {
+                return fmt.Errorf("cannot process value %v: %w", record.Value, err)
+        }
+
+        return out.Emit(ctx, core.StreamRecord[OUT]{
+                Type:      core.RecordData,
+                Key:       record.Key,
+                Timestamp: record.Timestamp,
+                Value:     outValue,
+        })
+}
+
+func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
+        return nil
+}
diff --git a/core/operator/operator.go b/core/operator/operator.go
new file mode 100644
index 0000000..0c588e5
--- /dev/null
+++ b/core/operator/operator.go
@@ -0,0 +1,16 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+)
+
+type Emitter[OUT any] interface {
+        Emit(ctx context.Context, record core.StreamRecord[OUT]) error
+}
+
+type Operator[IN, OUT any] interface {
+        Open(ctx context.Context) error
+        Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error
+        Close(ctx context.Context) error
+}
diff --git a/core/operator/stateful_map.go b/core/operator/stateful_map.go
new file mode 100644
index 0000000..295c656
--- /dev/null
+++ b/core/operator/stateful_map.go
@@ -0,0 +1,75 @@
+package operator
+
+import (
+        "context"
+        "fmt"
+
+        "flux/core"
+)
+
+// RichMapFunc is a stateful map on the user side: Open grabs the state handles
+// off ctx, Map runs per record with the backend already pointed at that
+// record's key.
+type RichMapFunc[IN, OUT any] interface {
+        Open(ctx context.Context) error
+        Map(ctx context.Context, in IN) (OUT, error)
+        Close(ctx context.Context) error
+}
+
+type StatefulMapOperator[IN, OUT any] struct {
+        userFunc RichMapFunc[IN, OUT]
+        backend  core.StateBackend
+}
+
+func NewStatefulMapOperator[IN, OUT any](fn RichMapFunc[IN, OUT]) *StatefulMapOperator[IN, OUT] {
+        return &StatefulMapOperator[IN, OUT]{userFunc: fn}
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Open(ctx context.Context) error {
+        // Get backend state from context via using context KV
+        be, ok := core.ExtractStateBackend(ctx)
+        if !ok || be == nil {
+                return fmt.Errorf("stateful map: no state backend in context")
+        }
+
+        s.backend = be
+        // The user func opens its ValueState off the same ctx.
+        return s.userFunc.Open(ctx)
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        if record.Type != core.RecordData {
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      record.Type,
+                        Timestamp: record.Timestamp,
+                        Key:       record.Key,
+                        BarrierID: record.BarrierID,
+                })
+        }
+
+        // Without a key every record would silently share the "" state slot.
+        if record.Key == "" {
+                return fmt.Errorf("stateful map: record %v has no key, keyBy it first", record.Value)
+        }
+
+        if err := s.backend.SetCurrentKey(record.Key); err != nil {
+                return fmt.Errorf("set current key %q: %w", record.Key, err)
+        }
+
+        // run the user transform
+        res, err := s.userFunc.Map(ctx, record.Value)
+        if err != nil {
+                return fmt.Errorf("cannot process value %v: %w", record.Value, err)
+        }
+
+        return out.Emit(ctx, core.StreamRecord[OUT]{
+                Type:      core.RecordData,
+                Timestamp: record.Timestamp,
+                Key:       record.Key,
+                Value:     res,
+        })
+}
+
+func (s *StatefulMapOperator[IN, OUT]) Close(ctx context.Context) error {
+        return s.userFunc.Close(ctx)
+}
diff --git a/core/operator/watermark.go b/core/operator/watermark.go
new file mode 100644
index 0000000..3d591e5
--- /dev/null
+++ b/core/operator/watermark.go
@@ -0,0 +1,61 @@
+package operator
+
+import (
+        "context"
+
+        "flux/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 {
+                if err := out.Emit(ctx, record); err != nil {
+                        return err
+                }
+                // 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
+}
diff --git a/core/operator/window.go b/core/operator/window.go
new file mode 100644
index 0000000..610f6e3
--- /dev/null
+++ b/core/operator/window.go
@@ -0,0 +1,110 @@
+package operator
+
+import (
+        "context"
+        "fmt"
+
+        "flux/core"
+)
+
+// WindowFunction is for user interface implements the behaviour
+type WindowFunction[IN, OUT any] interface {
+        Apply(ctx context.Context, key string, window core.Window, in []IN, out Emitter[OUT]) error
+}
+
+type WindowOperator[IN, OUT any] struct {
+        assigner    core.WindowAssigner[IN]
+        userfunc    WindowFunction[IN, OUT]
+        backend     core.StateBackend
+        timeSerivce core.InternalTimerService
+
+        windowState core.ValueState[[]IN]
+}
+
+func NewWindowOperator[IN, OUT any](assigner core.WindowAssigner[IN], userfunc WindowFunction[IN, OUT]) *WindowOperator[IN, OUT] {
+        return &WindowOperator[IN, OUT]{
+                assigner:    assigner,
+                userfunc:    userfunc,
+                timeSerivce: core.NewTimerService(),
+        }
+}
+
+func (o *WindowOperator[IN, OUT]) Open(ctx context.Context) error {
+        be, ok := core.ExtractStateBackend(ctx)
+        if !ok {
+                return fmt.Errorf("cannot extract state backend, got: %v", be)
+        }
+        o.backend = be
+        windowState, err := be.GetValueState(ctx, "window-elements")
+        if err != nil {
+                return fmt.Errorf("cannot get value state window-elements error: %w", err)
+        }
+
+        o.windowState = core.CastValueType[[]IN](windowState)
+
+        return nil
+}
+
+func (o *WindowOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        switch record.Type {
+        case core.RecordData:
+                // current windows state will hold all record in window
+                // assign and get list window
+                windows := o.assigner.AssignWindows(record.Value, record.Timestamp)
+                for _, w := range windows {
+                        o.backend.SetCurrentNamespace(w)
+                        o.backend.SetCurrentKey(record.Key)
+                        o.timeSerivce.Register(w.MaxTimestamp(), record.Key, w)
+
+                        currentWindowState, err := o.windowState.Value(ctx)
+                        if err != nil {
+                                return err
+                        }
+
+                        currentWindowState = append(currentWindowState, record.Value)
+
+                        o.windowState.Update(ctx, currentWindowState)
+                }
+
+                return nil
+        case core.RecordWatermark:
+                deadlines := o.timeSerivce.AdvanceWatermark(record.Timestamp)
+
+                for _, timer := range deadlines {
+                        o.backend.SetCurrentKey(timer.Key)
+                        o.backend.SetCurrentNamespace(timer.Namespace)
+
+                        currentWindowState, err := o.windowState.Value(ctx)
+                        if err != nil {
+                                return err
+                        }
+
+                        // NOTE: this is workaround create inline window, refactor later
+
+                        if err := o.userfunc.Apply(ctx, timer.Key, timer.Namespace.(core.Window), currentWindowState, out); err != nil {
+                                return err
+                        }
+
+                        if err := o.windowState.Clear(ctx); err != nil {
+                                return err
+                        }
+
+                }
+
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Timestamp: record.Timestamp,
+                        Type:      core.RecordWatermark,
+                })
+        default:
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        BarrierID: record.BarrierID,
+                        Timestamp: record.Timestamp,
+                        Key:       record.Key,
+                        Type:      record.Type,
+                })
+        }
+}
+
+func (o *WindowOperator[IN, OUT]) Close(ctx context.Context) error {
+        return nil
+}