From 0958e46c3651f2c8b6d76bcffbd8820063647d59 Mon Sep 17 00:00:00 2001 From: SangTran-127 Date: Sat, 8 Aug 2026 17:24:01 +0700 Subject: refactor: add context support to transport operators and runtime task management --- transport/channel.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) (limited to 'transport/channel.go') 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 } -- cgit v1.2.3