implement core transport and processing operators with receiver, emitter, filter, and map functionalities

5 files changed, 141 insertions, 0 deletions

diff --git a/core/filter.go b/core/filter.go
new file mode 100644
index 0000000..b882c0b
--- /dev/null
+++ b/core/filter.go
@@ -0,0 +1,31 @@
+package core
+
+type FilterFunc[IN any] func(in IN) bool
+
+type FilterOperator[IN any] struct {
+	userFunc FilterFunc[IN]
+}
+
+func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] {
+	return &FilterOperator[IN]{userFunc: userFunc}
+}
+
+func (f *FilterOperator[IN]) Open() error {
+	return nil
+}
+
+func (f *FilterOperator[IN]) Process(record StreamRecord[IN], out Emitter[IN]) error {
+	if record.Type != RecordData {
+		return out.Emit(record)
+	}
+
+	if f.userFunc(record.Value) {
+		return out.Emit(record)
+	}
+
+	return nil
+}
+
+func (f *FilterOperator[IN]) Close() error {
+	return nil
+}
diff --git a/core/map.go b/core/map.go
new file mode 100644
index 0000000..c011e8d
--- /dev/null
+++ b/core/map.go
@@ -0,0 +1,49 @@
+package core
+
+import "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 NewMapFuncOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
+	return &MapOperator[IN, OUT]{
+		userFunc: fn,
+	}
+}
+
+func (fo *MapOperator[IN, OUT]) Open() error {
+	return nil
+}
+
+func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT]) error {
+	if record.Type != RecordData {
+		//	 watermark and barrier
+		return out.Emit(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", record.Value)
+	}
+
+	return out.Emit(StreamRecord[OUT]{
+		Type:      RecordData,
+		Key:       record.Key,
+		Timestamp: record.Timestamp,
+		Value:     outValue,
+	})
+}
+
+func (fo *MapOperator[IN, OUT]) Close() error {
+	return nil
+}
diff --git a/core/operator.go b/core/operator.go
new file mode 100644
index 0000000..40d0ca6
--- /dev/null
+++ b/core/operator.go
@@ -0,0 +1,11 @@
+package core
+
+type Emitter[OUT any] interface {
+	Emit(record StreamRecord[OUT]) error
+}
+
+type Operator[IN, OUT any] interface {
+	Open() error
+	Process(record StreamRecord[IN], out Emitter[OUT]) error
+	Close() error
+}
diff --git a/transport/channel.go b/transport/channel.go
new file mode 100644
index 0000000..011f8ae
--- /dev/null
+++ b/transport/channel.go
@@ -0,0 +1,13 @@
+package transport
+
+import "goflink/core"
+
+type Receiver[T any] interface {
+	Receive() (core.StreamRecord[T], bool)
+	Close() error
+}
+
+type Emitter[T any] interface {
+	Emit(core.StreamRecord[T]) error
+	Close() error
+}
diff --git a/transport/local_channel.go b/transport/local_channel.go
new file mode 100644
index 0000000..4ac2c83
--- /dev/null
+++ b/transport/local_channel.go
@@ -0,0 +1,37 @@
+package transport
+
+import "goflink/core"
+
+type LocalReceiveChannel[T any] struct {
+	recordCh <-chan core.StreamRecord[T]
+}
+
+func NewLocalReceiveChannel[T any](ch <-chan core.StreamRecord[T]) *LocalReceiveChannel[T] {
+	return &LocalReceiveChannel[T]{
+		recordCh: ch,
+	}
+}
+
+func (c *LocalReceiveChannel[T]) Receive() (core.StreamRecord[T], bool) {
+	val, ok := <-c.recordCh
+	return val, ok
+}
+
+func (c *LocalReceiveChannel[T]) Close() error {
+	// The emitter will close
+	return nil
+}
+
+type LocalEmitChannel[T any] struct {
+	recordCh chan<- core.StreamRecord[T]
+}
+
+func (l LocalEmitChannel[T]) Emit(c core.StreamRecord[T]) error {
+	l.recordCh <- c
+	return nil
+}
+
+func (l LocalEmitChannel[T]) Close() error {
+	close(l.recordCh)
+	return nil
+}