From 4ca27aec303f54dc9ee670c67c2fefa80dd004a1 Mon Sep 17 00:00:00 2001 From: SangTran-127 Date: Sun, 9 Aug 2026 18:32:44 +0700 Subject: feat: add KeyByOperator and wordCounter for stateful processing --- core/operator/keyby.go | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 core/operator/keyby.go (limited to 'core/operator/keyby.go') diff --git a/core/operator/keyby.go b/core/operator/keyby.go new file mode 100644 index 0000000..1296df0 --- /dev/null +++ b/core/operator/keyby.go @@ -0,0 +1,32 @@ +package operator + +import ( + "context" + "goflink/core" +) + +type KeySelector[IN any] func(in IN) string +type KeyByOperator[IN any] struct { + selector KeySelector[IN] +} + +func NewKeyByOperator[IN any](selector KeySelector[IN]) *KeyByOperator[IN] { + return &KeyByOperator[IN]{selector: selector} +} + +func (k *KeyByOperator[IN]) Open(ctx context.Context) error { + return nil +} + +func (k *KeyByOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error { + if record.Type != core.RecordData { + return out.Emit(ctx, record) + } + + record.Key = k.selector(record.Value) + return out.Emit(ctx, record) +} + +func (k *KeyByOperator[IN]) Close(ctx context.Context) error { + return nil +} -- cgit v1.2.3