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
|
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()
}
|