Verified bug: fix memory_state clear and guard window state empty with implement main.go for test

4 files changed, 65 insertions, 33 deletions

diff --git a/core/operator/window.go b/core/operator/window.go
index 610f6e3..3f75d60 100644
--- a/core/operator/window.go
+++ b/core/operator/window.go
@@ -79,6 +79,10 @@ func (o *WindowOperator[IN, OUT]) Process(ctx context.Context, record core.Strea
                                 return err
                         }
 
+                        if len(currentWindowState) == 0 {
+                                continue
+                        }
+
                         // 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 {
diff --git a/main.go b/main.go
index 1b4a4be..91b96e3 100644
--- a/main.go
+++ b/main.go
@@ -19,24 +19,24 @@ type Transaction struct {
         Timestamp int64
 }
 
-type TransactionSummary[IN Transaction, OUT Transaction] struct {
+type TransactionSummary[IN Transaction, OUT string] struct {
         totalTransaction int
-        totalAmount      int
         average          int
 }
 
-func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out transport.Emitter[Transaction]) error {
+func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out operator.Emitter[string]) error {
+        totalAmount := 0
         for _, v := range in {
-                t.totalAmount += v.Amount
-                out.Emit(ctx, core.StreamRecord[Transaction]{
-                        Value: v,
-                        Type: core.RecordData,
-                        Timestamp: v.Timestamp,
-                })
+                totalAmount += v.Amount
         }
 
-        return nil
-}
+        msg := fmt.Sprintf("Khách: %s | Số đơn: %d | Tổng tiền: %d$", key, len(in), totalAmount)
+        
+        return out.Emit(ctx, core.StreamRecord[string]{
+                Value:     msg,
+                Type:      core.RecordData,
+                Timestamp: window.MaxTimestamp(),
+        })}
 
 func main() {
         // init
@@ -55,25 +55,45 @@ func main() {
                 {
                         User:      "Sang",
                         Amount:    4,
-                        Timestamp: 1200,
+                        Timestamp: 1500,
+                },
+                {
+                        User:      "Sang",
+                        Amount:    10,
+                        Timestamp: 2000,
+                },
+                {
+                        User:      "Sang",
+                        Amount:    12,
+                        Timestamp: 2500,
                 },
                 {
                         User:      "Sang",
                         Amount:    10,
-                        Timestamp: 1600,
+                        Timestamp: 3000,
                 },
         }
 
         ch1 := make(chan core.StreamRecord[Transaction], len(txList))
         ch2 := make(chan core.StreamRecord[Transaction], len(txList))
+        ch3 := make(chan core.StreamRecord[Transaction], len(txList))
+        ch4 := make(chan core.StreamRecord[string], len(txList))
 
         wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
                 return in.Timestamp
         })
 
+        ko := operator.NewKeyByOperator[Transaction](func(in Transaction) string {
+                return in.User
+        })
+
+        wop := operator.NewWindowOperator[Transaction, string](core.NewTumblingWindows[Transaction](1000), &TransactionSummary[Transaction, string]{})
+
         var wg sync.WaitGroup
 
-        runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
+        runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
+        runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch2), transport.NewLocalEmitChannel(ch3), ko)
+        runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch3), transport.NewLocalEmitChannel(ch4), wop)
 
         wg.Go(func() {
                 for i, tx := range txList {
@@ -84,13 +104,12 @@ func main() {
         })
 
         wg.Go(func() {
-                for v := range ch2 {
+                for v := range ch4 {
                         if v.Type == core.RecordData {
-                                fmt.Println("data: ", v.Value)
-                        } else {
-                                fmt.Println("watermark: ", v.Timestamp)
+                                fmt.Println("Total Amount after: ", v.Value)
                         }
                 }
         })
+
         wg.Wait()
 }
diff --git a/state/memory_backend.go b/state/memory_backend.go
index 6594657..f4043a1 100644
--- a/state/memory_backend.go
+++ b/state/memory_backend.go
@@ -2,6 +2,7 @@ package state
 
 import (
         "context"
+
         "flux/core"
 )
 
@@ -11,9 +12,9 @@ import (
 // This should be optimize with ZERO GC Scanning, Memory Alignment
 
 type MemoryStateBackend struct {
-        maxParallelism  int
-        currentKey      string
-        currentGroupKey int
+        maxParallelism   int
+        currentKey       string
+        currentGroupKey  int
         currentNamespace any
         // 4D structure StateName -> KeyGroup -> Key -> Namespace -> Value
         states map[string]map[int]map[string]map[any]any
@@ -36,18 +37,18 @@ func (ms *MemoryStateBackend) SetCurrentKey(key string) error {
         return nil
 }
 
-func (ms *MemoryStateBackend) GetValueState(ctx context.Context, key string) (core.ValueState[any], error) {
+func (ms *MemoryStateBackend) GetValueState(ctx context.Context, statename string) (core.ValueState[any], error) {
         // Assign if state not existed
-
-        if _, ok := ms.states[key]; !ok {
-                ms.states[key] = make(map[int]map[string]map[any]any)
+        
+        if _, ok := ms.states[statename]; !ok {
+                // create state name map if not existed 
+                ms.states[statename] = make(map[int]map[string]map[any]any)
         }
 
-        return &memoryStateValue{backend: ms, stateName: key}, nil
+        return &memoryStateValue{backend: ms, stateName: statename}, nil
 }
 
-func (ms *MemoryStateBackend) GetCurrentNamespace() any{
-
+func (ms *MemoryStateBackend) GetCurrentNamespace() any {
         return ms.currentNamespace
 }
 
diff --git a/state/memory_state.go b/state/memory_state.go
index 95c7318..97ecad1 100644
--- a/state/memory_state.go
+++ b/state/memory_state.go
@@ -1,7 +1,7 @@
 package state
 
 import (
-"context"
+        "context"
 )
 
 type memoryStateValue struct {
@@ -13,7 +13,7 @@ func (m *memoryStateValue) Value(ctx context.Context) (any, error) {
         kg := m.backend.currentGroupKey
         key := m.backend.currentKey
         ns := m.backend.currentNamespace
-        if kgMap, exist := m.backend.states[m.stateName][kg][key]; exist {
+        if kgMap, ok := m.backend.states[m.stateName][kg][key]; ok {
                 if v, hasKey := kgMap[ns]; hasKey {
                         return v, nil
                 }
@@ -26,10 +26,15 @@ func (m *memoryStateValue) Update(ctx context.Context, newVal any) (any, error)
         kg := m.backend.currentGroupKey
         key := m.backend.currentKey
         ns := m.backend.currentNamespace
-        if _, exists := m.backend.states[m.stateName][kg][key]; !exists {
+
+        if _, ok := m.backend.states[m.stateName][kg]; !ok {
                 m.backend.states[m.stateName][kg] = make(map[string]map[any]any)
         }
 
+        if _, ok := m.backend.states[m.stateName][kg][key]; !ok {
+                m.backend.states[m.stateName][kg][key] = make(map[any]any)
+        }
+
         m.backend.states[m.stateName][kg][key][ns] = newVal
         return newVal, nil
 }
@@ -37,8 +42,11 @@ func (m *memoryStateValue) Update(ctx context.Context, newVal any) (any, error)
 func (m *memoryStateValue) Clear(ctx context.Context) error {
         kg := m.backend.currentGroupKey
         key := m.backend.currentKey
-        if kgMap, exists := m.backend.states[m.stateName][kg]; exists {
-                delete(kgMap, key)
+        ns := m.backend.currentNamespace
+        if kgMap, ok := m.backend.states[m.stateName][kg]; ok {
+                if nsMap, ok := kgMap[key]; ok {
+                        delete(nsMap, ns)
+                }
         }
         return nil
 }