feat: add KeyByOperator and wordCounter for stateful processing

1 files changed, 0 insertions, 52 deletions

diff --git a/core/map.go b/core/map.go
deleted file mode 100644
index bbb8dcd..0000000
--- a/core/map.go
+++ /dev/null
@@ -1,52 +0,0 @@
-package core
-
-import (
-	"context"
-	"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 StreamRecord[IN], out Emitter[OUT]) error {
-	if record.Type != RecordData {
-		//	 watermark and barrier
-		return out.Emit(ctx, 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, StreamRecord[OUT]{
-		Type:      RecordData,
-		Key:       record.Key,
-		Timestamp: record.Timestamp,
-		Value:     outValue,
-	})
-}
-
-func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
-	return nil
-}