From 0958e46c3651f2c8b6d76bcffbd8820063647d59 Mon Sep 17 00:00:00 2001 From: SangTran-127 Date: Sat, 8 Aug 2026 17:24:01 +0700 Subject: refactor: add context support to transport operators and runtime task management --- core/map.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) (limited to 'core/map.go') 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 } -- cgit v1.2.3