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

1 files changed, 11 insertions, 8 deletions

diff --git a/core/map.go b/core/map.go
index c011e8d..bbb8dcd 100644
--- a/core/map.go
+++ b/core/map.go
@@ -1,6 +1,9 @@
 package core
 
-import "fmt"
+import (
+	"context"
+	"fmt"
+)
 
 // MapFunc is business logic for user implement
 type MapFunc[IN, OUT any] func(in IN) (OUT, error)
@@ -9,20 +12,20 @@ type MapOperator[IN, OUT any] struct {
 	userFunc MapFunc[IN, OUT]
 }
 
-func NewMapFuncOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
+func NewMapOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
 	return &MapOperator[IN, OUT]{
 		userFunc: fn,
 	}
 }
 
-func (fo *MapOperator[IN, OUT]) Open() error {
+func (fo *MapOperator[IN, OUT]) Open(ctx context.Context) error {
 	return nil
 }
 
-func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT]) error {
+func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record StreamRecord[IN], out Emitter[OUT]) error {
 	if record.Type != RecordData {
 		//	 watermark and barrier
-		return out.Emit(StreamRecord[OUT]{
+		return out.Emit(ctx, StreamRecord[OUT]{
 			Type:      record.Type,
 			Key:       record.Key,
 			Timestamp: record.Timestamp,
@@ -33,10 +36,10 @@ func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT
 	outValue, err := fo.userFunc(record.Value)
 
 	if err != nil {
-		return fmt.Errorf("cannot process value: %v", record.Value)
+		return fmt.Errorf("cannot process value %v: %w", record.Value, err)
 	}
 
-	return out.Emit(StreamRecord[OUT]{
+	return out.Emit(ctx, StreamRecord[OUT]{
 		Type:      RecordData,
 		Key:       record.Key,
 		Timestamp: record.Timestamp,
@@ -44,6 +47,6 @@ func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT
 	})
 }
 
-func (fo *MapOperator[IN, OUT]) Close() error {
+func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
 	return nil
 }