From 6631d8b00e8c5bafea15ea24415c5b55439193ee Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Mon, 31 Aug 2026 23:51:37 +0700
Subject: bug: fix memory_state clear and guard window state empty with
 implement main.go for test

---
 main.go | 55 +++++++++++++++++++++++++++++++++++++------------------
 1 file changed, 37 insertions(+), 18 deletions(-)

(limited to 'main.go')

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()
 }
-- 
cgit v1.3.1