1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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)
}
}
|