1 files changed, 32 insertions, 0 deletions

diff --git a/core/operator/keyby.go b/core/operator/keyby.go
new file mode 100644
index 0000000..56ef34e
--- /dev/null
+++ b/core/operator/keyby.go
@@ -0,0 +1,32 @@
+package operator
+
+import (
+        "context"
+        "flux/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
+}