feat: add KeyByOperator and wordCounter for stateful processing

1 files changed, 16 insertions, 0 deletions

diff --git a/core/operator/operator.go b/core/operator/operator.go
new file mode 100644
index 0000000..d4bf21b
--- /dev/null
+++ b/core/operator/operator.go
@@ -0,0 +1,16 @@
+package operator
+
+import (
+	"context"
+	"goflink/core"
+)
+
+type Emitter[OUT any] interface {
+	Emit(ctx context.Context, record core.StreamRecord[OUT]) error
+}
+
+type Operator[IN, OUT any] interface {
+	Open(ctx context.Context) error
+	Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error
+	Close(ctx context.Context) error
+}