package main

import (
	"context"
	"fmt"
	"strconv"

	"goflink/core"
	"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)
		}
	}()

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