Closed rvdwijngaard closed 1 year ago
Creating different ErrorSinks for each stage seems to work but it does not seem like a solid solution to me
package main
import (
"context"
"fmt"
"log"
"github.com/splunk/pipelines"
)
func main() {
ctx := context.Background()
input := pipelines.Chan([]string{"a", "b", "c", "d", "e", "f"})
ctx1, errs1 := pipelines.NewErrorSink(ctx)
ctx2, errs2 := pipelines.NewErrorSink(ctx)
stage1 := pipelines.Map(ctx1, input, func(item string) string {
if item == "c" {
errs1.Fatal(fmt.Errorf("%s caused an error at stage 1", item))
}
return item
})
stage2 := pipelines.Map(ctx2, stage1, func(item string) string {
if item == "b" {
errs2.Fatal(fmt.Errorf("%s caused an error at stage 2", item))
}
return item
})
_, err := pipelines.Drain(ctx2, stage2)
if err != nil {
log.Println(err.Error())
}
for _, err := range errs1.All() {
log.Println(err.Error())
}
for _, err := range errs2.All() {
log.Println(err.Error())
}
}
I have created #41 which fixes this error. Alternatively we could decide to provide ErrorSink
with a Close()
func which closes all open channels of the sink.
Thanks for the error report! I'm taking a look into the Close()
func idea today. Sorry about the delay.
See #43 for another approach which doesn't make any use of recover
. This adds a Close()
func like you suggested, and clarifies that the contract for ErrorSink only supports eventual consistency.
@kalexmills thanks for the fix. Your solution is better because it makes the contract clear.
This package looks promising. The concept of an ErrorSink is exactly what I was looking for. In my scenario I have a pipeline with multiple stages. Two stages need to make network calls. Some errors are fatal and the pipeline has to be canceled if one of these errors occur. When I use
ErrorSink.Fatal()
in multiple stages I get a panic:My test program looks like this:
Did I miss something or is this something which should work?