aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--core/filter.go12
-rw-r--r--core/map.go19
-rw-r--r--core/operator.go10
-rw-r--r--core/record.go12
-rw-r--r--main.go40
-rw-r--r--runtime/task.go66
-rw-r--r--runtime/task_test.go140
-rw-r--r--transport/channel.go16
-rw-r--r--transport/local_channel.go35
9 files changed, 314 insertions, 36 deletions
diff --git a/core/filter.go b/core/filter.go
index b882c0b..6c95fbc 100644
--- a/core/filter.go
+++ b/core/filter.go
@@ -1,5 +1,7 @@
package core
+import "context"
+
type FilterFunc[IN any] func(in IN) bool
type FilterOperator[IN any] struct {
@@ -10,22 +12,22 @@ func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] {
return &FilterOperator[IN]{userFunc: userFunc}
}
-func (f *FilterOperator[IN]) Open() error {
+func (f *FilterOperator[IN]) Open(ctx context.Context) error {
return nil
}
-func (f *FilterOperator[IN]) Process(record StreamRecord[IN], out Emitter[IN]) error {
+func (f *FilterOperator[IN]) Process(ctx context.Context, record StreamRecord[IN], out Emitter[IN]) error {
if record.Type != RecordData {
- return out.Emit(record)
+ return out.Emit(ctx, record)
}
if f.userFunc(record.Value) {
- return out.Emit(record)
+ return out.Emit(ctx, record)
}
return nil
}
-func (f *FilterOperator[IN]) Close() error {
+func (f *FilterOperator[IN]) Close(ctx context.Context) error {
return nil
}
diff --git a/core/map.go b/core/map.go
index c011e8d..bbb8dcd 100644
--- a/core/map.go
+++ b/core/map.go
@@ -1,6 +1,9 @@
package core
-import "fmt"
+import (
+ "context"
+ "fmt"
+)
// MapFunc is business logic for user implement
type MapFunc[IN, OUT any] func(in IN) (OUT, error)
@@ -9,20 +12,20 @@ type MapOperator[IN, OUT any] struct {
userFunc MapFunc[IN, OUT]
}
-func NewMapFuncOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
+func NewMapOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] {
return &MapOperator[IN, OUT]{
userFunc: fn,
}
}
-func (fo *MapOperator[IN, OUT]) Open() error {
+func (fo *MapOperator[IN, OUT]) Open(ctx context.Context) error {
return nil
}
-func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT]) error {
+func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record StreamRecord[IN], out Emitter[OUT]) error {
if record.Type != RecordData {
// watermark and barrier
- return out.Emit(StreamRecord[OUT]{
+ return out.Emit(ctx, StreamRecord[OUT]{
Type: record.Type,
Key: record.Key,
Timestamp: record.Timestamp,
@@ -33,10 +36,10 @@ func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT
outValue, err := fo.userFunc(record.Value)
if err != nil {
- return fmt.Errorf("cannot process value: %v", record.Value)
+ return fmt.Errorf("cannot process value %v: %w", record.Value, err)
}
- return out.Emit(StreamRecord[OUT]{
+ return out.Emit(ctx, StreamRecord[OUT]{
Type: RecordData,
Key: record.Key,
Timestamp: record.Timestamp,
@@ -44,6 +47,6 @@ func (fo *MapOperator[IN, OUT]) Process(record StreamRecord[IN], out Emitter[OUT
})
}
-func (fo *MapOperator[IN, OUT]) Close() error {
+func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error {
return nil
}
diff --git a/core/operator.go b/core/operator.go
index 40d0ca6..7785d44 100644
--- a/core/operator.go
+++ b/core/operator.go
@@ -1,11 +1,13 @@
package core
+import "context"
+
type Emitter[OUT any] interface {
- Emit(record StreamRecord[OUT]) error
+ Emit(ctx context.Context, record StreamRecord[OUT]) error
}
type Operator[IN, OUT any] interface {
- Open() error
- Process(record StreamRecord[IN], out Emitter[OUT]) error
- Close() error
+ Open(ctx context.Context) error
+ Process(ctx context.Context, record StreamRecord[IN], out Emitter[OUT]) error
+ Close(ctx context.Context) error
}
diff --git a/core/record.go b/core/record.go
index cb160b2..e18e58d 100644
--- a/core/record.go
+++ b/core/record.go
@@ -16,8 +16,8 @@ type StreamRecord[T any] struct {
Type RecordType
}
-func NewDataRecord[T any](key string, val T, timestamp int64) *StreamRecord[T] {
- return &StreamRecord[T]{
+func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] {
+ return StreamRecord[T]{
Value: val,
Key: key,
Timestamp: timestamp,
@@ -25,15 +25,15 @@ func NewDataRecord[T any](key string, val T, timestamp int64) *StreamRecord[T] {
}
}
-func NewWatermarkRecord[T any](timestamp int64) *StreamRecord[T] {
- return &StreamRecord[T]{
+func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] {
+ return StreamRecord[T]{
Type: RecordWatermark,
Timestamp: timestamp,
}
}
-func NewBarrierRecord[T any](barrierID uint64) *StreamRecord[T] {
- return &StreamRecord[T]{
+func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] {
+ return StreamRecord[T]{
Type: RecordBarrier,
BarrierID: barrierID,
}
diff --git a/main.go b/main.go
index 25792c1..806c1ed 100644
--- a/main.go
+++ b/main.go
@@ -1,13 +1,49 @@
package main
import (
+ "context"
"fmt"
+ "strconv"
+
"goflink/core"
- "unsafe"
+ "goflink/runtime"
+ "goflink/transport"
)
func main() {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := make(chan core.StreamRecord[int])
+ mid := make(chan core.StreamRecord[int])
+ sink := make(chan core.StreamRecord[string])
+
+ // x*10, then render as text
+ mapped := runtime.RunTask(ctx,
+ transport.NewLocalReceiveChannel(source),
+ transport.NewLocalEmitChannel(mid),
+ core.NewMapOperator(func(in int) (int, error) { return in * 10, nil }),
+ )
+ printed := runtime.RunTask(ctx,
+ transport.NewLocalReceiveChannel(mid),
+ transport.NewLocalEmitChannel(sink),
+ core.NewMapOperator(func(in int) (string, error) { return "v=" + strconv.Itoa(in), nil }),
+ )
+
+ go func() {
+ defer close(source)
+ for i := 1; i <= 5; i++ {
+ source <- core.NewDataRecord("k", i, 0)
+ }
+ }()
- fmt.Printf("GoodRecord Size: %d bytes\n", unsafe.Sizeof(core.StreamRecord[int]{}))
+ for record := range sink {
+ fmt.Println(record.Value)
+ }
+ for _, errCh := range []<-chan error{mapped, printed} {
+ if err := <-errCh; err != nil {
+ fmt.Println("task failed:", err)
+ }
+ }
}
diff --git a/runtime/task.go b/runtime/task.go
new file mode 100644
index 0000000..3c887bb
--- /dev/null
+++ b/runtime/task.go
@@ -0,0 +1,66 @@
+package runtime
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "goflink/core"
+ "goflink/transport"
+)
+
+// RunTask runs a stream task in the background. The returned channel yields the
+// first error that stopped it (context.Canceled when ctx ends early) and is
+// closed when the task is done, so callers can join on it.
+func RunTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], operator core.Operator[IN, OUT]) <-chan error {
+ done := make(chan error, 1)
+
+ go func() {
+ defer close(done)
+ if err := runTask(ctx, input, output, operator); err != nil {
+ done <- err
+ }
+ }()
+
+ return done
+}
+
+func runTask[IN, OUT any](ctx context.Context, input transport.Receiver[IN], output transport.Emitter[OUT], operator core.Operator[IN, OUT]) (err error) {
+ // ponytail: assumes one writer per output channel. Fan-in needs a refcount
+ // or a dedicated closer, otherwise the second Close panics.
+ defer func() {
+ if cerr := output.Close(); cerr != nil && err == nil {
+ err = fmt.Errorf("close output: %w", cerr)
+ }
+ }()
+
+ if err := operator.Open(ctx); err != nil {
+ return fmt.Errorf("open operator: %w", err)
+ }
+ defer func() {
+ // ctx is usually already cancelled by the time we get here — that is
+ // why the task stopped. Close still needs a live window to flush, so
+ // give it one that keeps ctx's values but drops the cancellation.
+ closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+ defer cancel()
+
+ if cerr := operator.Close(closeCtx); cerr != nil && err == nil {
+ err = fmt.Errorf("close operator: %w", cerr)
+ }
+ }()
+
+ for {
+ record, ok := input.Receive(ctx)
+ if !ok {
+ break
+ }
+
+ if err := operator.Process(ctx, record, output); err != nil {
+ return fmt.Errorf("process record: %w", err)
+ }
+ }
+
+ // Receive also returns !ok on cancellation, so report that as an error
+ // rather than a clean drain.
+ return ctx.Err()
+}
diff --git a/runtime/task_test.go b/runtime/task_test.go
new file mode 100644
index 0000000..f27a5eb
--- /dev/null
+++ b/runtime/task_test.go
@@ -0,0 +1,140 @@
+package runtime
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "goflink/core"
+ "goflink/transport"
+)
+
+// map(x*10) -> filter(>20), end to end.
+func TestPipeline(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ source := make(chan core.StreamRecord[int])
+ mid := make(chan core.StreamRecord[int])
+ sink := make(chan core.StreamRecord[int])
+
+ mapped := RunTask(ctx,
+ transport.NewLocalReceiveChannel(source),
+ transport.NewLocalEmitChannel(mid),
+ core.NewMapOperator(func(in int) (int, error) { return in * 10, nil }),
+ )
+ filtered := RunTask(ctx,
+ transport.NewLocalReceiveChannel(mid),
+ transport.NewLocalEmitChannel(sink),
+ core.NewFilterOperator(func(in int) bool { return in > 20 }),
+ )
+
+ go func() {
+ defer close(source)
+ for i := 1; i <= 5; i++ {
+ source <- core.NewDataRecord("k", i, int64(i))
+ }
+ }()
+
+ var got []int
+ for record := range sink {
+ got = append(got, record.Value)
+ }
+
+ want := []int{30, 40, 50}
+ if len(got) != len(want) {
+ t.Fatalf("got %v, want %v", got, want)
+ }
+ for i := range want {
+ if got[i] != want[i] {
+ t.Fatalf("got %v, want %v", got, want)
+ }
+ }
+
+ for _, errCh := range []<-chan error{mapped, filtered} {
+ if err := <-errCh; err != nil {
+ t.Fatalf("task failed: %v", err)
+ }
+ }
+}
+
+// A cancelled ctx must unblock a task parked in Receive.
+func TestCancelStopsTask(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+
+ source := make(chan core.StreamRecord[int])
+ sink := make(chan core.StreamRecord[int], 1)
+
+ done := RunTask(ctx,
+ transport.NewLocalReceiveChannel(source),
+ transport.NewLocalEmitChannel(sink),
+ core.NewMapOperator(func(in int) (int, error) { return in, nil }),
+ )
+
+ cancel()
+
+ if err := <-done; !errors.Is(err, context.Canceled) {
+ t.Fatalf("got %v, want context.Canceled", err)
+ }
+}
+
+// A user function error must reach the caller with its cause intact.
+func TestProcessErrorPropagates(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ boom := errors.New("boom")
+ source := make(chan core.StreamRecord[int], 1)
+ sink := make(chan core.StreamRecord[int], 1)
+
+ done := RunTask(ctx,
+ transport.NewLocalReceiveChannel(source),
+ transport.NewLocalEmitChannel(sink),
+ core.NewMapOperator(func(in int) (int, error) { return 0, boom }),
+ )
+
+ source <- core.NewDataRecord("k", 1, 0)
+
+ if err := <-done; !errors.Is(err, boom) {
+ t.Fatalf("got %v, want %v", err, boom)
+ }
+}
+
+// closeProbe records the ctx.Err() its Close saw.
+type closeProbe struct {
+ closeCtxErr error
+}
+
+func (o *closeProbe) Open(ctx context.Context) error { return nil }
+
+func (o *closeProbe) Process(ctx context.Context, r core.StreamRecord[int], out core.Emitter[int]) error {
+ return out.Emit(ctx, r)
+}
+
+func (o *closeProbe) Close(ctx context.Context) error {
+ o.closeCtxErr = ctx.Err()
+ return nil
+}
+
+// Close must get a live ctx even when the task was killed by cancellation,
+// otherwise an operator that flushes on shutdown loses whatever it buffered.
+func TestCloseGetsLiveContext(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+
+ source := make(chan core.StreamRecord[int])
+ sink := make(chan core.StreamRecord[int], 1)
+
+ probe := &closeProbe{}
+ done := RunTask(ctx,
+ transport.NewLocalReceiveChannel(source),
+ transport.NewLocalEmitChannel(sink),
+ probe,
+ )
+
+ cancel()
+ <-done // Close has already run by the time this channel closes
+
+ if probe.closeCtxErr != nil {
+ t.Fatalf("Close got a dead ctx (%v), so it cannot flush", probe.closeCtxErr)
+ }
+}
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
}
diff --git a/transport/local_channel.go b/transport/local_channel.go
index 4ac2c83..e3e4cba 100644
--- a/transport/local_channel.go
+++ b/transport/local_channel.go
@@ -1,6 +1,10 @@
package transport
-import "goflink/core"
+import (
+ "context"
+
+ "goflink/core"
+)
type LocalReceiveChannel[T any] struct {
recordCh <-chan core.StreamRecord[T]
@@ -12,9 +16,14 @@ func NewLocalReceiveChannel[T any](ch <-chan core.StreamRecord[T]) *LocalReceive
}
}
-func (c *LocalReceiveChannel[T]) Receive() (core.StreamRecord[T], bool) {
- val, ok := <-c.recordCh
- return val, ok
+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 {
@@ -26,12 +35,22 @@ 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 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 {
+func (l *LocalEmitChannel[T]) Close() error {
close(l.recordCh)
return nil
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-18 18:38:59 +0000