update(GroupBy): add GroupByOperator

doc.go

@@ -132,7 +132,7 @@
 // The following operations may cause concurrent behavior due to [Context]
 // cancellation:
 //   - CombineLatest operators (due to use of [Synchronize]);
-//   - [Connect] (due to use of [Multicast]);
+//   - [Connect] and [GroupBy] (due to use of [Multicast]);
 //   - Merge operators (due to use of [Synchronize]);
 //   - [Multicast] and other relatives (due to use of [Context.AfterFunc] and
 //     [Synchronize]);

groupby.go

@@ -3,21 +3,39 @@ package rx
 // GroupBy groups the values emitted by the source [Observable] according to
 // a specified criterion, and emits these grouped values as Pairs, one [Pair]
 // per group.
-func GroupBy[T any, K comparable](
-	keySelector func(v T) K,
-	groupFactory func() Subject[T],
-) Operator[T, Pair[K, Observable[T]]] {
-	return NewOperator(
-		func(source Observable[T]) Observable[Pair[K, Observable[T]]] {
-			return groupByObservable[T, K]{source, keySelector, groupFactory}.Subscribe
+func GroupBy[T any, K comparable](keySelector func(v T) K) GroupByOperator[T, K] {
+	return GroupByOperator[T, K]{
+		ts: groupByConfig[T, K]{
+			keySelector:   keySelector,
+			groupSupplier: Multicast[T],
 		},
-	)
+	}
+}
+
+type groupByConfig[T any, K comparable] struct {
+	keySelector   func(T) K
+	groupSupplier func() Subject[T]
+}
+
+// GroupByOperator is an [Operator] type for [GroupBy].
+type GroupByOperator[T any, K comparable] struct {
+	ts groupByConfig[T, K]
+}
+
+// WithGroupSupplier sets GroupSupplier option to a given value.
+func (op GroupByOperator[T, K]) WithGroupSupplier(groupSupplier func() Subject[T]) GroupByOperator[T, K] {
+	op.ts.groupSupplier = groupSupplier
+	return op
+}
+
+// Apply implements the [Operator] interface.
+func (op GroupByOperator[T, K]) Apply(source Observable[T]) Observable[Pair[K, Observable[T]]] {
+	return groupByObservable[T, K]{source, op.ts}.Subscribe
 }
 
 type groupByObservable[T any, K comparable] struct {
-	source       Observable[T]
-	keySelector  func(T) K
-	groupFactory func() Subject[T]
+	source Observable[T]
+	groupByConfig[T, K]
 }
 
 func (ob groupByObservable[T, K]) Subscribe(c Context, o Observer[Pair[K, Observable[T]]]) {
@@ -30,7 +48,7 @@ func (ob groupByObservable[T, K]) Subscribe(c Context, o Observer[Pair[K, Observ
 			group, exists := groups[key]
 
 			if !exists {
-				g := ob.groupFactory()
+				g := ob.groupSupplier()
 				group = g.Observer
 				groups[key] = group
 				o.Next(NewPair(key, g.Observable))

groupby_test.go

@@ -17,8 +17,7 @@ func TestGroupBy(t *testing.T) {
 
 	group := rx.GroupBy(
 		func(v string) string { return v },
-		rx.MulticastBufferAll[string],
-	)
+	).WithGroupSupplier(rx.MulticastBufferAll[string])
 
 	count := rx.ConcatMap(
 		func(g rx.Pair[string, rx.Observable[string]]) rx.Observable[rx.Pair[string, int]] {
@@ -56,8 +55,7 @@ func TestGroupBy(t *testing.T) {
 			source,
 			rx.GroupBy(
 				func(v string) string { panic(ErrTest) },
-				rx.MulticastBufferAll[string],
-			),
+			).WithGroupSupplier(rx.MulticastBufferAll[string]),
 			count,
 			tostring,
 		),
@@ -67,8 +65,7 @@ func TestGroupBy(t *testing.T) {
 			source,
 			rx.GroupBy(
 				func(v string) string { return v },
-				func() rx.Subject[string] { panic(ErrTest) },
-			),
+			).WithGroupSupplier(func() rx.Subject[string] { panic(ErrTest) }),
 			count,
 			tostring,
 		),