1 files changed, 96 insertions, 0 deletions

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