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

1 files changed, 37 insertions, 0 deletions

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
+}