diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/filter.go | 12 | ||||
| -rw-r--r-- | core/map.go | 19 | ||||
| -rw-r--r-- | core/operator.go | 10 | ||||
| -rw-r--r-- | core/record.go | 12 |
4 files changed, 30 insertions, 23 deletions
diff --git a/core/filter.go b/core/filter.go index b882c0b..6c95fbc 100644 --- a/core/filter.go +++ b/core/filter.go @@ -1,5 +1,7 @@ package core +import "context" + type FilterFunc[IN any] func(in IN) bool type FilterOperator[IN any] struct { @@ -10,22 +12,22 @@ func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] { return &FilterOperator[IN]{userFunc: userFunc} } -func (f *FilterOperator[IN]) Open() error { +func (f *FilterOperator[IN]) Open(ctx context.Context) error { return nil } -func (f *FilterOperator[IN]) Process(record StreamRecord[IN], out Emitter[IN]) error { +func (f *FilterOperator[IN]) Process(ctx context.Context, record StreamRecord[IN], out Emitter[IN]) error { if record.Type != RecordData { - return out.Emit(record) + return out.Emit(ctx, record) } if f.userFunc(record.Value) { - return out.Emit(record) + return out.Emit(ctx, record) } return nil } -func (f *FilterOperator[IN]) Close() error { +func (f *FilterOperator[IN]) Close(ctx context.Context) error { return nil } 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 } diff --git a/core/operator.go b/core/operator.go index 40d0ca6..7785d44 100644 --- a/core/operator.go +++ b/core/operator.go @@ -1,11 +1,13 @@ package core +import "context" + type Emitter[OUT any] interface { - Emit(record StreamRecord[OUT]) error + Emit(ctx context.Context, record StreamRecord[OUT]) error } type Operator[IN, OUT any] interface { - Open() error - Process(record StreamRecord[IN], out Emitter[OUT]) error - Close() error + Open(ctx context.Context) error + Process(ctx context.Context, record StreamRecord[IN], out Emitter[OUT]) error + Close(ctx context.Context) error } diff --git a/core/record.go b/core/record.go index cb160b2..e18e58d 100644 --- a/core/record.go +++ b/core/record.go @@ -16,8 +16,8 @@ type StreamRecord[T any] struct { Type RecordType } -func NewDataRecord[T any](key string, val T, timestamp int64) *StreamRecord[T] { - return &StreamRecord[T]{ +func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] { + return StreamRecord[T]{ Value: val, Key: key, Timestamp: timestamp, @@ -25,15 +25,15 @@ func NewDataRecord[T any](key string, val T, timestamp int64) *StreamRecord[T] { } } -func NewWatermarkRecord[T any](timestamp int64) *StreamRecord[T] { - return &StreamRecord[T]{ +func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] { + return StreamRecord[T]{ Type: RecordWatermark, Timestamp: timestamp, } } -func NewBarrierRecord[T any](barrierID uint64) *StreamRecord[T] { - return &StreamRecord[T]{ +func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] { + return StreamRecord[T]{ Type: RecordBarrier, BarrierID: barrierID, } |