From 91b1c642601efbbcc5af9931e087f68b2c72fc3d Mon Sep 17 00:00:00 2001
From: SangTran-127 <tranquangsang12.7@gmail.com>
Date: Thu, 13 Aug 2026 23:42:53 +0700
Subject: feat: add on timer & watermark

---
 main.go | 94 -----------------------------------------------------------------
 1 file changed, 94 deletions(-)

(limited to 'main.go')

diff --git a/main.go b/main.go
index 1d86d66..da29a2c 100644
--- a/main.go
+++ b/main.go
@@ -1,98 +1,4 @@
 package main
 
-import (
-	"context"
-	"errors"
-	"fmt"
-
-	"goflink/core"
-	"goflink/core/operator"
-	"goflink/runtime"
-	"goflink/state"
-	"goflink/transport"
-)
-
-// wordCounter counts how many times each key has been seen, in keyed state.
-type wordCounter struct {
-	count core.ValueState[any]
-}
-
-func (w *wordCounter) Open(ctx context.Context) error {
-	be, ok := core.ExtractStateBackend(ctx)
-	if !ok {
-		return errors.New("no state backend in context")
-	}
-	s, err := be.GetValueState(ctx, "word-count")
-	if err != nil {
-		return err
-	}
-	w.count = s
-	return nil
-}
-
-func (w *wordCounter) Map(ctx context.Context, in string) (int, error) {
-	v, err := w.count.Value(ctx)
-	if err != nil {
-		return 0, err
-	}
-	n, _ := v.(int) // unseen key -> nil -> 0
-	n++
-	_, err = w.count.Update(ctx, n)
-	return n, err
-}
-
-func (w *wordCounter) Close(ctx context.Context) error { return nil }
-
 func main() {
-
-	/*
-			Bây giờ bạn đã có đầy đủ đồ chơi: MemoryStateBackend, KeyByOperator, StatefulMapOperator.
-
-		Trong file main.go, bạn hãy thiết lập một Pipeline có dạng: Source ➡️ KeyBy (Trích xuất Tên) ➡️ StatefulMap (Đếm số lần Tên xuất hiện) ➡️ Sink
-
-		Yêu cầu Logic của hàm StatefulMap:
-
-		Trong hàm Open, bạn hãy khởi tạo một ValueState[int] tên là "word-count".
-		Trong hàm xử lý (User Func), bạn hãy đọc giá trị cũ lên, cộng thêm 1, lưu lại vào State, và trả về con số mới đó.
-		Hãy chạy thử xem nếu bạn truyền vào ["Sang", "Tran", "Sang", "Sang"], nó có in ra 1, 1, 2, 3 không nhé. Nếu thành công, bạn đã tự tay chế tạo xong một Streaming Engine Stateful đích thực!
-	*/
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	// ponytail: one backend for the whole job. Each stateful task needs its own
-	// once tasks run on separate key-group shards.
-	ctx = core.InjectStateBackend(ctx, state.NewMemoryStateBackend(core.DefaultMaxParallelism))
-
-	source := make(chan core.StreamRecord[string])
-	mid := make(chan core.StreamRecord[string])
-	sink := make(chan core.StreamRecord[int])
-
-	// keyBy(name) -> count occurrences per name
-	keyed := runtime.RunTask(ctx,
-		transport.NewLocalReceiveChannel(source),
-		transport.NewLocalEmitChannel(mid),
-		operator.NewKeyByOperator(func(in string) string { return in }),
-	)
-	counted := runtime.RunTask(ctx,
-		transport.NewLocalReceiveChannel(mid),
-		transport.NewLocalEmitChannel(sink),
-		operator.NewStatefulMapOperator[string, int](&wordCounter{}),
-	)
-
-	go func() {
-		defer close(source)
-		for _, name := range []string{"Sang", "Tran", "Sang", "Sang"} {
-			source <- core.NewDataRecord("", name, 0)
-		}
-	}()
-
-	for record := range sink {
-		fmt.Println(record.Key, record.Value)
-	}
-
-	for _, errCh := range []<-chan error{keyed, counted} {
-		if err := <-errCh; err != nil {
-			fmt.Println("task failed:", err)
-		}
-	}
 }
-- 
cgit v1.2.3