1 files changed, 53 insertions, 0 deletions

diff --git a/core/operator/map.go b/core/operator/map.go
new file mode 100644
index 0000000..2165c25
--- /dev/null
+++ b/core/operator/map.go
@@ -0,0 +1,53 @@
+package operator
+
+import (
+        "context"
+        "flux/core"
+        "fmt"
+)
+
+// MapFunc is business logic for user implement
+type MapFunc[IN, OUT any] func(in IN) (OUT, error)
+
+type MapOperator[IN, OUT any] struct {
+        userFunc MapFunc[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(ctx context.Context) error {
+        return nil
+}
+
+func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error {
+        if record.Type != core.RecordData {
+                //       watermark and barrier
+                return out.Emit(ctx, core.StreamRecord[OUT]{
+                        Type:      record.Type,
+                        Key:       record.Key,
+                        Timestamp: record.Timestamp,
+                        BarrierID: record.BarrierID,
+                })
+        }
+
+        outValue, err := fo.userFunc(record.Value)
+
+        if err != nil {
+                return fmt.Errorf("cannot process value %v: %w", record.Value, err)
+        }
+
+        return out.Emit(ctx, core.StreamRecord[OUT]{
+                Type:      core.RecordData,
+                Key:       record.Key,
+                Timestamp: record.Timestamp,
+                Value:     outValue,
+        })
+}
+
+func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
+        return nil
+}