RxGo. How can I stop an observable stream?

user_4935400 0
Viewed 104

I'm just a beginner in learning the GO language.

I'm familiar with RxJava, RxJs, etc.

While studying rxgo, there is a part that I do not understand.

Why is 3 being printed even though an error has occurred? Why didn't the StopOnError option apply?

import (
    "context"
    "errors"
    "fmt"

    "github.com/reactivex/rxgo/v2"
)

func exam() { 
    observable := rxgo.Create([]rxgo.Producer{
        func(ctx context.Context, next chan<- rxgo.Item) {
            next <- rxgo.Of(1)
            next <- rxgo.Of(2)
            next <- rxgo.Error(errors.New("unknown"))
            next <- rxgo.Of(3)
        },
    }, rxgo.WithErrorStrategy(rxgo.StopOnError))

    for v := range observable.Observe() {
        fmt.Printf("CHECK %+v\n", v)
    }
}

OUTPUT

CHECK {V:1 E:<nil>}
CHECK {V:2 E:<nil>}
CHECK {V:<nil> E:unknown}
CHECK {V:3 E:<nil>}
rxgo go
1 Answers

Looks like library expects client to perform some transformation on Observable to trigger strategy. I tried Map but based on source code, it is embedded in other methods.

It is not very intuitive, and I did not find if that behavior is documented.

import (
    "context"
    "errors"
    "fmt"

    "github.com/reactivex/rxgo/v2"
)

func main() {
    errs := rxgo.Just(
        rxgo.Of(1),
        errors.New("foo"),
        rxgo.Of(2),
        rxgo.Of(3),
    )()
    for v := range errs.Observe(rxgo.WithErrorStrategy(rxgo.StopOnError)) {
        fmt.Printf("CHECK %+v %v\n", v, v.Error())
    }

    errs = rxgo.Just(
        rxgo.Of(1),
        errors.New("foo"),
        rxgo.Of(2),
        rxgo.Of(3),
    )().Map(func(ctx context.Context, i interface{}) (interface{}, error) {
        return i, nil
    })
    for v := range errs.Observe(rxgo.WithErrorStrategy(rxgo.StopOnError)) {
        fmt.Printf("CHECK MAP %+v %v\n", v, v.Error())
    }
}

https://go.dev/play/p/n2Cr-p6pUC5

It is actually can be a bug in library. rxgo.StopOnError is a default strategy and there is an example for overriding it:

https://github.com/ReactiveX/RxGo/blob/6d27cca9572a67c6c423d4d4d41afd8edf5b1d72/doc/errors.md

Problem with that example that changing rxgo.Continue to rxgo.StopOnError does not change anything:


    errs := rxgo.Just(
        errors.New("foo"),
        errors.New("bar"),
        errors.New("baz"),
    )().Errors(rxgo.WithErrorStrategy(rxgo.StopOnError))
    fmt.Println(errs)

prints

[foo bar baz]

https://go.dev/play/p/swFwTo-q7c2

user_14203320
Related