1 files changed, 13 insertions, 3 deletions
diff --git a/transport/channel.go b/transport/channel.go
index 011f8ae..1db6ea8 100644
--- a/transport/channel.go
+++ b/transport/channel.go
@@ -1,13 +1,23 @@
 package transport
 
-import "goflink/core"
+import (
+	"context"
+
+	"goflink/core"
+)
 
 type Receiver[T any] interface {
-	Receive() (core.StreamRecord[T], bool)
+	// Receive returns the next record, or ok=false once the stream is done or
+	// ctx is cancelled.
+	Receive(ctx context.Context) (core.StreamRecord[T], bool)
 	Close() error
 }
 
+// Emitter mirrors core.Emitter plus Close. It is declared separately because
+// core cannot import transport without an import cycle.
 type Emitter[T any] interface {
-	Emit(core.StreamRecord[T]) error
+	// Emit blocks until the record is handed off, or returns ctx.Err() if ctx
+	// is cancelled first.
+	Emit(ctx context.Context, record core.StreamRecord[T]) error
 	Close() error
 }