feat: add KeyByOperator and wordCounter for stateful processing

1 files changed, 7 insertions, 7 deletions

diff --git a/runtime/task.go b/runtime/task.go
index 3c887bb..b3d1caa 100644
--- a/runtime/task.go
+++ b/runtime/task.go
@@ -5,19 +5,19 @@ import (
 	"fmt"
 	"time"
 
-	"goflink/core"
+	"goflink/core/operator"
 	"goflink/transport"
 )
 
 // RunTask runs a stream task in the background. The returned channel yields the
 // first error that stopped it (context.Canceled when ctx ends early) and is
 // closed when the task is done, so callers can join on it.
-func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], operator core.Operator[IN, OUT]) <-chan error {
+func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) <-chan error {
 	done := make(chan error, 1)
 
 	go func() {
 		defer close(done)
-		if err := runTask(ctx, input, output, operator); err != nil {
+		if err := runTask(ctx, input, output, op); err != nil {
 			done <- err
 		}
 	}()
@@ -25,7 +25,7 @@ func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], out
 	return done
 }
 
-func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], operator core.Operator[IN, OUT]) (err error) {
+func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], op operator.Operator[IN, OUT]) (err error) {
 	// ponytail: assumes one writer per output channel. Fan-in needs a refcount
 	// or a dedicated closer, otherwise the second Close panics.
 	defer func() {
@@ -34,7 +34,7 @@ func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], out
 		}
 	}()
 
-	if err := operator.Open(ctx); err != nil {
+	if err := op.Open(ctx); err != nil {
 		return fmt.Errorf("open operator: %w", err)
 	}
 	defer func() {
@@ -44,7 +44,7 @@ func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], out
 		closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
 		defer cancel()
 
-		if cerr := operator.Close(closeCtx); cerr != nil && err == nil {
+		if cerr := op.Close(closeCtx); cerr != nil && err == nil {
 			err = fmt.Errorf("close operator: %w", cerr)
 		}
 	}()
@@ -55,7 +55,7 @@ func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], out
 			break
 		}
 
-		if err := operator.Process(ctx, record, output); err != nil {
+		if err := op.Process(ctx, record, output); err != nil {
 			return fmt.Errorf("process record: %w", err)
 		}
 	}