refactor: add context support to transport operators and runtime task management

1 files changed, 38 insertions, 2 deletions

diff --git a/main.go b/main.go
index 25792c1..806c1ed 100644
--- a/main.go
+++ b/main.go
@@ -1,13 +1,49 @@
 package main
 
 import (
+	"context"
 	"fmt"
+	"strconv"
+
 	"goflink/core"
-	"unsafe"
+	"goflink/runtime"
+	"goflink/transport"
 )
 
 func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	source := make(chan core.StreamRecord[int])
+	mid := make(chan core.StreamRecord[int])
+	sink := make(chan core.StreamRecord[string])
+
+	// x*10, then render as text
+	mapped := runtime.RunTask(ctx,
+		transport.NewLocalReceiveChannel(source),
+		transport.NewLocalEmitChannel(mid),
+		core.NewMapOperator(func(in int) (int, error) { return in * 10, nil }),
+	)
+	printed := runtime.RunTask(ctx,
+		transport.NewLocalReceiveChannel(mid),
+		transport.NewLocalEmitChannel(sink),
+		core.NewMapOperator(func(in int) (string, error) { return "v=" + strconv.Itoa(in), nil }),
+	)
+
+	go func() {
+		defer close(source)
+		for i := 1; i <= 5; i++ {
+			source <- core.NewDataRecord("k", i, 0)
+		}
+	}()
 
-	fmt.Printf("GoodRecord Size: %d bytes\n", unsafe.Sizeof(core.StreamRecord[int]{}))
+	for record := range sink {
+		fmt.Println(record.Value)
+	}
 
+	for _, errCh := range []<-chan error{mapped, printed} {
+		if err := <-errCh; err != nil {
+			fmt.Println("task failed:", err)
+		}
+	}
 }