rename Serialize function to Synchronize

combinelatest2.go

@@ -11,7 +11,7 @@ func CombineLatest2[T1, T2, R any](
 	mapping func(v1 T1, v2 T2) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState2[T1, T2]
 

combinelatest3.go

@@ -12,7 +12,7 @@ func CombineLatest3[T1, T2, T3, R any](
 	mapping func(v1 T1, v2 T2, v3 T3) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState3[T1, T2, T3]
 

combinelatest4.go

@@ -13,7 +13,7 @@ func CombineLatest4[T1, T2, T3, T4, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState4[T1, T2, T3, T4]
 

combinelatest5.go

@@ -14,7 +14,7 @@ func CombineLatest5[T1, T2, T3, T4, T5, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState5[T1, T2, T3, T4, T5]
 

combinelatest6.go

@@ -15,7 +15,7 @@ func CombineLatest6[T1, T2, T3, T4, T5, T6, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState6[T1, T2, T3, T4, T5, T6]
 

combinelatest7.go

@@ -16,7 +16,7 @@ func CombineLatest7[T1, T2, T3, T4, T5, T6, T7, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState7[T1, T2, T3, T4, T5, T6, T7]
 

combinelatest8.go

@@ -17,7 +17,7 @@ func CombineLatest8[T1, T2, T3, T4, T5, T6, T7, T8, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState8[T1, T2, T3, T4, T5, T6, T7, T8]
 

combinelatest9.go

@@ -18,7 +18,7 @@ func CombineLatest9[T1, T2, T3, T4, T5, T6, T7, T8, T9, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8, v9 T9) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s combineLatestState9[T1, T2, T3, T4, T5, T6, T7, T8, T9]
 

doc.go

@@ -131,17 +131,17 @@
 //
 // The following operations may cause concurrent behavior due to [Context]
 // cancellation:
-//   - CombineLatest operators (due to use of [Serialize]);
+//   - CombineLatest operators (due to use of [Synchronize]);
 //   - [Connect] (due to use of [Multicast]);
-//   - Merge operators (due to use of [Serialize]);
+//   - Merge operators (due to use of [Synchronize]);
 //   - [Multicast] and other relatives (due to use of [Context.AfterFunc] and
-//     [Serialize]);
+//     [Synchronize]);
 //   - [Never] (due to use of [Context.AfterFunc]);
-//   - [Serialize] (due to use of [Context.AfterFunc]);
 //   - [Share] (due to use of [Context.AfterFunc] and [Multicast]);
+//   - [Synchronize] (due to use of [Context.AfterFunc]);
 //   - [Unicast] and other relatives (due to use of [Context.AfterFunc]);
-//   - WithLatestFrom operators (due to use of [Serialize]);
-//   - ZipWithBuffering operators (due to use of [Serialize]).
+//   - WithLatestFrom operators (due to use of [Synchronize]);
+//   - ZipWithBuffering operators (due to use of [Synchronize]).
 //
 // Since [Context] cancellations are very common in this library, and that
 // a [Context] cancellation usually results in a [Stop] notification, emitted

merge.go

@@ -34,7 +34,7 @@ type mergeWithObservable[T any] struct {
 }
 
 func (ob mergeWithObservable[T]) Subscribe(c Context, o Observer[T]) {
-	c, o = Serialize(c, o)
+	c, o = Synchronize(c, o)
 
 	var num atomic.Uint32
 
@@ -156,7 +156,7 @@ func (ob mergeMapObservable[T, R]) Subscribe(c Context, o Observer[R]) {
 		return
 	}
 
-	c, o = Serialize(c, o)
+	c, o = Synchronize(c, o)
 
 	var x struct {
 		mu sync.Mutex
@@ -254,7 +254,7 @@ func (ob mergeMapObservable[T, R]) Subscribe(c Context, o Observer[R]) {
 }
 
 func (ob mergeMapObservable[T, R]) SubscribeWithBuffering(c Context, o Observer[R]) {
-	c, o = Serialize(c, o)
+	c, o = Synchronize(c, o)
 
 	var x struct {
 		mu sync.Mutex

multicast.go

@@ -150,7 +150,7 @@ func (m *multicast[T]) Subscribe(c Context, o Observer[T]) {
 
 	lastn := m.lastn
 	if lastn.Kind == 0 {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		o := o
 		m.mobs.Add(&o)

multicast_test.go

@@ -19,7 +19,7 @@ func TestMulticast(t *testing.T) {
 
 		subscribeThenComplete := rx.NewObservable(
 			func(c rx.Context, o rx.Observer[string]) {
-				c, o = rx.Serialize(c, o)
+				c, o = rx.Synchronize(c, o)
 				m.Subscribe(c, o)
 				o.Complete()
 			},

observer.go

@@ -67,6 +67,17 @@ func (o Observer[T]) DoOnTermination(f func()) Observer[T] {
 	}
 }
 
+// Synchronize returns an [Observer] that passes incoming emissions to o
+// in a mutually exclusive way.
+// Synchronize also returns a copy of c that will be cancelled when o is
+// about to receive a notification of [Complete], [Error] or [Stop].
+func Synchronize[T any](c Context, o Observer[T]) (Context, Observer[T]) {
+	c, cancel := c.WithCancel()
+	u := new(unicast[T])
+	u.Subscribe(c, o.DoOnTermination(cancel))
+	return c, u.Emit
+}
+
 // WithRuntimeFinalizer creates an Observer with a runtime finalizer set to
 // run o.Stop(ErrFinalized) in a goroutine.
 // o must be safe for concurrent use.

withlatestfrom1.go

@@ -22,7 +22,7 @@ func withLatestFrom2[T1, T2, R any](
 	mapping func(v1 T1, v2 T2) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState2[T1, T2]
 

withlatestfrom2.go

@@ -24,7 +24,7 @@ func withLatestFrom3[T1, T2, T3, R any](
 	mapping func(v1 T1, v2 T2, v3 T3) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState3[T1, T2, T3]
 

withlatestfrom3.go

@@ -26,7 +26,7 @@ func withLatestFrom4[T1, T2, T3, T4, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState4[T1, T2, T3, T4]
 

withlatestfrom4.go

@@ -28,7 +28,7 @@ func withLatestFrom5[T1, T2, T3, T4, T5, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState5[T1, T2, T3, T4, T5]
 

withlatestfrom5.go

@@ -30,7 +30,7 @@ func withLatestFrom6[T1, T2, T3, T4, T5, T6, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState6[T1, T2, T3, T4, T5, T6]
 

withlatestfrom6.go

@@ -32,7 +32,7 @@ func withLatestFrom7[T1, T2, T3, T4, T5, T6, T7, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState7[T1, T2, T3, T4, T5, T6, T7]
 

withlatestfrom7.go

@@ -34,7 +34,7 @@ func withLatestFrom8[T1, T2, T3, T4, T5, T6, T7, T8, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState8[T1, T2, T3, T4, T5, T6, T7, T8]
 

withlatestfrom8.go

@@ -36,7 +36,7 @@ func withLatestFrom9[T1, T2, T3, T4, T5, T6, T7, T8, T9, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8, v9 T9) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s withLatestFromState9[T1, T2, T3, T4, T5, T6, T7, T8, T9]
 

zipwithbuffering2.go

@@ -18,7 +18,7 @@ func ZipWithBuffering2[T1, T2, R any](
 	mapping func(v1 T1, v2 T2) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState2[T1, T2]
 

zipwithbuffering3.go

@@ -19,7 +19,7 @@ func ZipWithBuffering3[T1, T2, T3, R any](
 	mapping func(v1 T1, v2 T2, v3 T3) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState3[T1, T2, T3]
 

zipwithbuffering4.go

@@ -20,7 +20,7 @@ func ZipWithBuffering4[T1, T2, T3, T4, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState4[T1, T2, T3, T4]
 

zipwithbuffering5.go

@@ -21,7 +21,7 @@ func ZipWithBuffering5[T1, T2, T3, T4, T5, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState5[T1, T2, T3, T4, T5]
 

zipwithbuffering6.go

@@ -22,7 +22,7 @@ func ZipWithBuffering6[T1, T2, T3, T4, T5, T6, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState6[T1, T2, T3, T4, T5, T6]
 

zipwithbuffering7.go

@@ -23,7 +23,7 @@ func ZipWithBuffering7[T1, T2, T3, T4, T5, T6, T7, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState7[T1, T2, T3, T4, T5, T6, T7]
 

zipwithbuffering8.go

@@ -24,7 +24,7 @@ func ZipWithBuffering8[T1, T2, T3, T4, T5, T6, T7, T8, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState8[T1, T2, T3, T4, T5, T6, T7, T8]
 

zipwithbuffering9.go

@@ -25,7 +25,7 @@ func ZipWithBuffering9[T1, T2, T3, T4, T5, T6, T7, T8, T9, R any](
 	mapping func(v1 T1, v2 T2, v3 T3, v4 T4, v5 T5, v6 T6, v7 T7, v8 T8, v9 T9) R,
 ) Observable[R] {
 	return func(c Context, o Observer[R]) {
-		c, o = Serialize(c, o)
+		c, o = Synchronize(c, o)
 
 		var s zipState9[T1, T2, T3, T4, T5, T6, T7, T8, T9]