2 files changed, 50 insertions, 0 deletions
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
+}