feat(rx): add backoff-retry operator (#156)

Author: plastikfan

.vscode/settings.json

@@ -7,6 +7,7 @@
     "Assistable",
     "binaryheap",
     "bodyclose",
+    "cenkalti",
     "cmds",
     "coverpkg",
     "coverprofile",

go.mod

@@ -22,6 +22,7 @@ require (
 )
 
 require (
+	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/emirpasic/gods v1.18.1
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect

go.sum

@@ -2,6 +2,8 @@ github.com/BurntSushi/toml v1.3.2 h1:o7IhLm0Msx3BaB+n3Ag7L8EVlByGnpq14C4YWiu/gL8
 github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
 github.com/avfs/avfs v0.33.0 h1:5WQXbUbr6VS7aani39ZN2Vrd/s3wLnyih1Sc4ExWTxs=
 github.com/avfs/avfs v0.33.0/go.mod h1:Q59flcFRYe9KYkNMfrLUJney3yeKGQpcWRyxsDBW7vI=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

rx/observable-operator-backoff-retry_test.go

@@ -0,0 +1,74 @@
+package rx_test
+
+import (
+	"context"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/fortytw2/leaktest"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
+
+	"github.com/snivilised/lorax/rx"
+)
+
+var _ = Describe("Observable operator", func() {
+	Context("Backoff Retry", func() {
+		Context("principle", func() {
+			It("🧪 should: succeed after retry within max retries", func() {
+				// rxgo: Test_Observable_BackOffRetry
+				defer leaktest.Check(GinkgoT())()
+
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+
+				i := 0
+				backOffCfg := backoff.NewExponentialBackOff()
+				backOffCfg.InitialInterval = time.Nanosecond
+				obs := rx.Defer([]rx.Producer[int]{func(_ context.Context, next chan<- rx.Item[int]) {
+					next <- rx.Of(1)
+					next <- rx.Of(2)
+					if i == 2 {
+						next <- rx.Of(3)
+					} else {
+						i++
+						next <- rx.Error[int](errFoo)
+					}
+				}}).BackOffRetry(backoff.WithMaxRetries(backOffCfg, 3))
+				rx.Assert(ctx, obs,
+					rx.HasItems[int]{
+						Expected: []int{1, 2, 1, 2, 1, 2, 3},
+					},
+					rx.HasNoError[int]{},
+				)
+			})
+		})
+
+		Context("Errors", func() {
+			Context("given: foo", func() {
+				It("🧪 should: fail after max retries exceeded", func() {
+					// rxgo: Test_Observable_BackOffRetry_Error
+					defer leaktest.Check(GinkgoT())()
+
+					ctx, cancel := context.WithCancel(context.Background())
+					defer cancel()
+
+					backOffCfg := backoff.NewExponentialBackOff()
+					backOffCfg.InitialInterval = time.Nanosecond
+					obs := rx.Defer([]rx.Producer[int]{func(_ context.Context, next chan<- rx.Item[int]) {
+						next <- rx.Of(1)
+						next <- rx.Of(2)
+						next <- rx.Error[int](errFoo)
+					}}).BackOffRetry(backoff.WithMaxRetries(backOffCfg, 3))
+					rx.Assert(ctx, obs,
+						rx.HasItems[int]{
+							Expected: []int{1, 2, 1, 2, 1, 2, 1, 2},
+						},
+						rx.HasError[int]{
+							Expected: []error{errFoo},
+						},
+					)
+				})
+			})
+		})
+	})
+})

rx/observable-operator.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+
+	"github.com/cenkalti/backoff/v4"
 )
 
 func isZero[T any](limit T) bool {
@@ -134,6 +136,52 @@ func (op *averageOperator[T]) gatherNext(_ context.Context, item Item[T],
 	panic("averageOperator.gatherNext NOT-IMPL")
 }
 
+// BackOffRetry implements a backoff retry if a source Observable sends an error,
+// resubscribe to it in the hopes that it will complete without error.
+// Cannot be run in parallel.
+func (o *ObservableImpl[T]) BackOffRetry(backOffCfg backoff.BackOff, opts ...Option[T]) Observable[T] {
+	option := parseOptions(opts...)
+	next := option.buildChannel()
+	ctx := option.buildContext(o.parent)
+	f := func() error {
+		observe := o.Observe(opts...)
+
+		for {
+			select {
+			case <-ctx.Done():
+				close(next)
+
+				return nil
+			case i, ok := <-observe:
+				if !ok {
+					return nil
+				}
+
+				if i.IsError() {
+					return i.E
+				}
+
+				i.SendContext(ctx, next)
+			}
+		}
+	}
+
+	go func() {
+		if err := backoff.Retry(f, backOffCfg); err != nil {
+			Error[T](err).SendContext(ctx, next)
+			close(next)
+
+			return
+		}
+
+		close(next)
+	}()
+
+	return &ObservableImpl[T]{
+		iterable: newChannelIterable(next),
+	}
+}
+
 // Connect instructs a connectable Observable to begin emitting items to its subscribers.
 func (o *ObservableImpl[T]) Connect(ctx context.Context) (context.Context, Disposable) {
 	ctx, cancel := context.WithCancel(ctx)

rx/observable.go

@@ -5,13 +5,15 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/emirpasic/gods/trees/binaryheap"
 )
 
 type Observable[T any] interface {
 	Iterable[T]
 	All(predicate Predicate[T], opts ...Option[T]) Single[T]
 	Average(calc Calculator[T], opts ...Option[T]) Single[T]
+	BackOffRetry(backOffCfg backoff.BackOff, opts ...Option[T]) Observable[T]
 	Connect(ctx context.Context) (context.Context, Disposable)
 
 	Max(comparator Comparator[T], initLimit InitLimit[T], opts ...Option[T]) OptionalSingle[T]

rx/operator-skeleton_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 var _ = Describe("Observable operator", func() {
-	Context("${{OPERATOR-NAME}}", func() {
+	XContext("${{OPERATOR-NAME}}", func() {
 		Context("principle", func() {
 			// success path
 			It("🧪 should: ", func() {