From 96c07bab1086510a55a004e9a1b90c30bc723a5c Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 29 Aug 2026 01:08:28 +0700
Subject: feat: init base draft support window

---
 main.go | 96 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 96 insertions(+)
 create mode 100644 main.go

(limited to 'main.go')

diff --git a/main.go b/main.go
new file mode 100644
index 0000000..1b4a4be
--- /dev/null
+++ b/main.go
@@ -0,0 +1,96 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+
+	"flux/core"
+	"flux/core/operator"
+	"flux/runtime"
+	"flux/state"
+	"flux/transport"
+)
+
+type Transaction struct {
+	User      string
+	Amount    int
+	Timestamp int64
+}
+
+type TransactionSummary[IN Transaction, OUT Transaction] 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 {
+	for _, v := range in {
+		t.totalAmount += v.Amount
+		out.Emit(ctx, core.StreamRecord[Transaction]{
+			Value: v,
+			Type: core.RecordData,
+			Timestamp: v.Timestamp,
+		})
+	}
+
+	return nil
+}
+
+func main() {
+	// init
+	ctx := context.Background()
+
+	backend := state.NewMemoryStateBackend(1)
+
+	ctx = core.InjectStateBackend(ctx, backend)
+
+	txList := []Transaction{
+		{
+			User:      "Sang",
+			Amount:    1,
+			Timestamp: 1000,
+		},
+		{
+			User:      "Sang",
+			Amount:    4,
+			Timestamp: 1200,
+		},
+		{
+			User:      "Sang",
+			Amount:    10,
+			Timestamp: 1600,
+		},
+	}
+
+	ch1 := make(chan core.StreamRecord[Transaction], len(txList))
+	ch2 := make(chan core.StreamRecord[Transaction], len(txList))
+
+	wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
+		return in.Timestamp
+	})
+
+	var wg sync.WaitGroup
+
+	runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
+
+	wg.Go(func() {
+		for i, tx := range txList {
+			ch1 <- core.NewDataRecord(strconv.Itoa(i), tx, tx.Timestamp)
+		}
+
+		close(ch1)
+	})
+
+	wg.Go(func() {
+		for v := range ch2 {
+			if v.Type == core.RecordData {
+				fmt.Println("data: ", v.Value)
+			} else {
+				fmt.Println("watermark: ", v.Timestamp)
+			}
+		}
+	})
+	wg.Wait()
+}
-- 
cgit v1.3.1