From 96c07bab1086510a55a004e9a1b90c30bc723a5c Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 29 Aug 2026 01:08:28 +0700
Subject: feat: init base draft support window

---
 transport/local_channel.go | 56 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 56 insertions(+)
 create mode 100644 transport/local_channel.go

(limited to 'transport/local_channel.go')

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
+}
-- 
cgit v1.3.1