1 files changed, 56 insertions, 0 deletions

diff --git a/transport/local_channel.go b/transport/local_channel.go
new file mode 100644
index 0000000..2fac029
--- /dev/null
+++ b/transport/local_channel.go
@@ -0,0 +1,56 @@
+package transport
+
+import (
+        "context"
+
+        "flux/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(ctx context.Context) (core.StreamRecord[T], bool) {
+        select {
+        case val, ok := <-c.recordCh:
+                return val, ok
+        case <-ctx.Done():
+                var zero core.StreamRecord[T]
+                return zero, false
+        }
+}
+
+func (c *LocalReceiveChannel[T]) Close() error {
+        // The emitter will close
+        return nil
+}
+
+type LocalEmitChannel[T any] struct {
+        recordCh chan<- core.StreamRecord[T]
+}
+
+func NewLocalEmitChannel[T any](ch chan<- core.StreamRecord[T]) *LocalEmitChannel[T] {
+        return &LocalEmitChannel[T]{
+                recordCh: ch,
+        }
+}
+
+func (l *LocalEmitChannel[T]) Emit(ctx context.Context, c core.StreamRecord[T]) error {
+        select {
+        case l.recordCh <- c:
+                return nil
+        case <-ctx.Done():
+                return ctx.Err()
+        }
+}
+
+func (l *LocalEmitChannel[T]) Close() error {
+        close(l.recordCh)
+        return nil
+}