feat(rx): add find operator (#174)

rx/observable-operator-find_test.go

@@ -0,0 +1,59 @@
+package rx_test
+
+import (
+	"context"
+
+	"github.com/fortytw2/leaktest"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
+	"github.com/snivilised/lorax/rx"
+)
+
+var _ = Describe("Observable operator", func() {
+	Context("Find", func() {
+		Context("not empty", func() {
+			When("is present", func() {
+				It("🧪 should: return requested item", func() {
+					// rxgo: Test_Observable_Find_NotEmpty
+					defer leaktest.Check(GinkgoT())()
+
+					ctx, cancel := context.WithCancel(context.Background())
+					defer cancel()
+					obs := testObservable[int](ctx, 1, 2, 3).Find(func(item rx.Item[int]) bool {
+						return item.V == 2
+					})
+					rx.Assert(ctx, obs, rx.HasItem[int]{
+						Expected: 2,
+					})
+				})
+			})
+
+			When("is not present", func() {
+				It("🧪 should: return nothing", func() {
+					defer leaktest.Check(GinkgoT())()
+
+					ctx, cancel := context.WithCancel(context.Background())
+					defer cancel()
+					obs := testObservable[int](ctx, 1, 2, 3).Find(func(item rx.Item[int]) bool {
+						return item.V == 99
+					})
+					rx.Assert(ctx, obs, rx.IsEmpty[int]{})
+				})
+			})
+		})
+
+		When("empty", func() {
+			It("🧪 should: return nothing", func() {
+				// rxgo: Test_Observable_Find_Empty
+				defer leaktest.Check(GinkgoT())()
+
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+
+				obs := rx.Empty[int]().Find(func(_ rx.Item[int]) bool {
+					return true
+				})
+				rx.Assert(ctx, obs, rx.IsEmpty[int]{})
+			})
+		})
+	})
+})

rx/observable-operator.go

@@ -706,6 +706,46 @@ func (op *filterOperator[T]) gatherNext(_ context.Context, _ Item[T],
 ) {
 }
 
+// Find emits the first item passing a predicate then complete.
+func (o *ObservableImpl[T]) Find(find Predicate[T], opts ...Option[T]) OptionalSingle[T] {
+	const (
+		forceSeq     = true
+		bypassGather = true
+	)
+
+	return optionalSingle(o.parent, o, func() operator[T] {
+		return &findOperator[T]{
+			find: find,
+		}
+	}, true, true, opts...)
+}
+
+type findOperator[T any] struct {
+	find Predicate[T]
+}
+
+func (op *findOperator[T]) next(ctx context.Context, item Item[T],
+	dst chan<- Item[T], operatorOptions operatorOptions[T],
+) {
+	if op.find(item) {
+		item.SendContext(ctx, dst)
+		operatorOptions.stop()
+	}
+}
+
+func (op *findOperator[T]) err(ctx context.Context, item Item[T],
+	dst chan<- Item[T], operatorOptions operatorOptions[T],
+) {
+	defaultErrorFuncOperator(ctx, item, dst, operatorOptions)
+}
+
+func (op *findOperator[T]) end(_ context.Context, _ chan<- Item[T]) {
+}
+
+func (op *findOperator[T]) gatherNext(_ context.Context, _ Item[T],
+	_ chan<- Item[T], _ operatorOptions[T]) {
+}
+
 // !!!
 
 // Max determines and emits the maximum-valued item emitted by an Observable according to a comparator.

rx/observable.go

@@ -27,6 +27,7 @@ type Observable[T any] interface {
 	Error(opts ...Option[T]) error
 	Errors(opts ...Option[T]) []error
 	Filter(apply Predicate[T], opts ...Option[T]) Observable[T]
+	Find(find Predicate[T], opts ...Option[T]) OptionalSingle[T]
 	Max(comparator Comparator[T], initLimit InitLimit[T], opts ...Option[T]) OptionalSingle[T]
 	Map(apply Func[T], opts ...Option[T]) Observable[T]
 	Min(comparator Comparator[T], initLimit InitLimit[T], opts ...Option[T]) OptionalSingle[T]