1 files changed, 37 insertions, 18 deletions

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()
 }