update(Share): add Context option

b97tsk · Aug 12, 2024 · 4f19df7 · 4f19df7
1 parent ad6b9b8
commit 4f19df7
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 13 deletions.
diff --git a/share.go b/share.go
@@ -9,10 +9,10 @@ import (
 // [Observable]. When subscribed multiple times, it guarantees that only one
 // subscription is made to the source at the same time. When all subscribers
 // have unsubscribed it will unsubscribe from the source.
-func Share[T any](c Context) ShareOperator[T] {
+func Share[T any]() ShareOperator[T] {
 	return ShareOperator[T]{
 		ts: shareConfig[T]{
-			context:   c,
+			context:   NewBackgroundContext(),
 			connector: Multicast[T],
 		},
 	}
@@ -34,6 +34,12 @@ func (op ShareOperator[T]) WithConnector(connector func() Subject[T]) ShareOpera
 	return op
 }
 
+// WithContext sets Context option to a given value.
+func (op ShareOperator[T]) WithContext(c Context) ShareOperator[T] {
+	op.ts.context = c
+	return op
+}
+
 // Apply implements the [Operator] interface.
 func (op ShareOperator[T]) Apply(source Observable[T]) Observable[T] {
 	ob := &shareObservable[T]{

diff --git a/share_test.go b/share_test.go
@@ -11,15 +11,13 @@ import (
 func TestShare(t *testing.T) {
 	t.Parallel()
 
-	ctx := rx.NewBackgroundContext()
-
 	NewTestSuite[int](t).Case(
 		func() rx.Observable[int] {
 			ob := rx.Pipe3(
 				rx.Ticker(Step(3)),
 				rx.Scan(-1, func(i int, _ time.Time) int { return i + 1 }),
 				rx.Take[int](4),
-				rx.Share[int](ctx),
+				rx.Share[int](),
 			)
 			return rx.Merge(
 				ob,
@@ -34,7 +32,7 @@ func TestShare(t *testing.T) {
 			ob := rx.Pipe3(
 				rx.Ticker(Step(3)),
 				rx.Scan(-1, func(i int, _ time.Time) int { return i + 1 }),
-				rx.Share[int](ctx),
+				rx.Share[int](),
 				rx.Take[int](4),
 			)
 			return rx.Merge(
@@ -51,11 +49,11 @@ func TestShare(t *testing.T) {
 				rx.Ticker(Step(3)),
 				rx.Scan(-1, func(i int, _ time.Time) int { return i + 1 }),
 				rx.Take[int](4),
-				rx.Share[int](ctx).WithConnector(
+				rx.Share[int]().WithConnector(
 					func() rx.Subject[int] {
 						return rx.MulticastBuffer[int](1)
 					},
-				),
+				).WithContext(rx.NewBackgroundContext()),
 			)
 			return rx.Merge(
 				ob,
@@ -70,11 +68,11 @@ func TestShare(t *testing.T) {
 			ob := rx.Pipe3(
 				rx.Ticker(Step(3)),
 				rx.Scan(-1, func(i int, _ time.Time) int { return i + 1 }),
-				rx.Share[int](ctx).WithConnector(
+				rx.Share[int]().WithConnector(
 					func() rx.Subject[int] {
 						return rx.MulticastBuffer[int](1)
 					},
-				),
+				).WithContext(rx.NewBackgroundContext()),
 				rx.Take[int](4),
 			)
 			return rx.Merge(
@@ -88,7 +86,7 @@ func TestShare(t *testing.T) {
 	).Case(
 		rx.Pipe1(
 			rx.Oops[int]("should not happen"),
-			rx.Share[int](ctx).WithConnector(
+			rx.Share[int]().WithConnector(
 				func() rx.Subject[int] {
 					return rx.Subject[int]{
 						Observable: rx.Throw[int](ErrTest),
@@ -101,7 +99,7 @@ func TestShare(t *testing.T) {
 	).Case(
 		rx.Pipe1(
 			rx.Oops[int]("should not happen"),
-			rx.Share[int](ctx).WithConnector(
+			rx.Share[int]().WithConnector(
 				func() rx.Subject[int] {
 					return rx.Subject[int]{
 						Observable: rx.Oops[int](ErrTest),
@@ -114,7 +112,7 @@ func TestShare(t *testing.T) {
 	).Case(
 		rx.Pipe1(
 			rx.Oops[int]("should not happen"),
-			rx.Share[int](ctx).WithConnector(func() rx.Subject[int] { panic(ErrTest) }),
+			rx.Share[int]().WithConnector(func() rx.Subject[int] { panic(ErrTest) }),
 		),
 		rx.ErrOops, ErrTest,
 	)