Revert "update(*): use c.Cause() rather than c.Err()"

This reverts commit 4071f7f.

blocking.go

@@ -6,7 +6,7 @@ package rx
 // the error.
 //
 // The cancellation of parent will cause BlockingFirst to immediately return
-// the zero value of T and parent.Cause().
+// the zero value of T and parent.Err().
 func (ob Observable[T]) BlockingFirst(parent Context) (v T, err error) {
 	res := Error[T](ErrEmpty)
 	c, cancel := parent.WithCancel()
@@ -36,7 +36,7 @@ func (ob Observable[T]) BlockingFirst(parent Context) (v T, err error) {
 	select {
 	default:
 	case <-parent.Done():
-		return v, parent.Cause()
+		return v, parent.Err()
 	}
 
 	switch res.Kind {
@@ -69,7 +69,7 @@ func (ob Observable[T]) BlockingFirstOrElse(parent Context, def T) T {
 // the error.
 //
 // The cancellation of parent will cause BlockingLast to immediately return
-// the zero value of T and parent.Cause().
+// the zero value of T and parent.Err().
 func (ob Observable[T]) BlockingLast(parent Context) (v T, err error) {
 	res := Error[T](ErrEmpty)
 	c, cancel := parent.WithCancel()
@@ -91,7 +91,7 @@ func (ob Observable[T]) BlockingLast(parent Context) (v T, err error) {
 	select {
 	default:
 	case <-parent.Done():
-		return v, parent.Cause()
+		return v, parent.Err()
 	}
 
 	switch res.Kind {
@@ -124,7 +124,7 @@ func (ob Observable[T]) BlockingLastOrElse(parent Context, def T) T {
 // error, it returns the zero value of T and the error.
 //
 // The cancellation of parent will cause BlockingSingle to immediately return
-// the zero value of T and parent.Cause().
+// the zero value of T and parent.Err().
 func (ob Observable[T]) BlockingSingle(parent Context) (v T, err error) {
 	res := Error[T](ErrEmpty)
 	c, cancel := parent.WithCancel()
@@ -159,7 +159,7 @@ func (ob Observable[T]) BlockingSingle(parent Context) (v T, err error) {
 	select {
 	default:
 	case <-parent.Done():
-		return v, parent.Cause()
+		return v, parent.Err()
 	}
 
 	switch res.Kind {
@@ -177,7 +177,7 @@ func (ob Observable[T]) BlockingSingle(parent Context) (v T, err error) {
 // otherwise, it returns the emitted error.
 //
 // The cancellation of parent will cause BlockingSubscribe to immediately
-// return parent.Cause().
+// return parent.Err().
 func (ob Observable[T]) BlockingSubscribe(parent Context, o Observer[T]) error {
 	var res Notification[T]
 
@@ -197,7 +197,7 @@ func (ob Observable[T]) BlockingSubscribe(parent Context, o Observer[T]) error {
 	select {
 	default:
 	case <-parent.Done():
-		return parent.Cause()
+		return parent.Err()
 	}
 
 	switch res.Kind {

catch.go

@@ -17,7 +17,7 @@ func Catch[T any](selector func(err error) Observable[T]) Operator[T, T] {
 						select {
 						default:
 						case <-c.Done():
-							o.Error(c.Cause())
+							o.Error(c.Err())
 							return
 						}
 
@@ -50,7 +50,7 @@ func OnErrorResumeWith[T any](ob Observable[T]) Operator[T, T] {
 						select {
 						default:
 						case <-c.Done():
-							o.Error(c.Cause())
+							o.Error(c.Err())
 							return
 						}
 
@@ -82,7 +82,7 @@ func OnErrorComplete[T any]() Operator[T, T] {
 						select {
 						default:
 						case <-c.Done():
-							o.Error(c.Cause())
+							o.Error(c.Err())
 							return
 						}
 

concat.go

@@ -49,7 +49,7 @@ func (ob concatWithObservable[T]) Subscribe(c Context, o Observer[T]) {
 		select {
 		default:
 		case <-done:
-			o.Error(c.Cause())
+			o.Error(c.Err())
 			return
 		}
 
@@ -209,7 +209,7 @@ func (ob concatMapObservable[T, R]) SubscribeWithBuffering(c Context, o Observer
 					select {
 					default:
 					case <-w.Done():
-						n = Error[R](w.Cause())
+						n = Error[R](w.Err())
 					}
 				}
 

context.go

@@ -41,6 +41,11 @@ func (c Context) Done() <-chan struct{} {
 	return c.Context.Done()
 }
 
+// Err returns c.Context.Err().
+func (c Context) Err() error {
+	return c.Context.Err()
+}
+
 // Wait runs c.WaitGroup.Wait().
 // If c.WaitGroup is not set, Wait panics.
 func (c Context) Wait() {

delay.go

@@ -66,7 +66,7 @@ func (ob delayObservable[T]) Subscribe(c Context, o Observer[T]) {
 						x.Queue.Unlock()
 
 						if old != sentinel {
-							o.Error(w.Cause())
+							o.Error(w.Err())
 						}
 
 						return

empty.go

@@ -9,12 +9,12 @@ func Empty[T any]() Observable[T] {
 
 // Never returns an Observable that never emits anything, except
 // when a context cancellation is detected, emits an error notification
-// of whatever [Context.Cause] returns.
+// of whatever that context reports.
 func Never[T any]() Observable[T] {
 	return func(c Context, o Observer[T]) {
 		if c.Done() != nil {
 			c.AfterFunc(func() {
-				o.Error(c.Cause())
+				o.Error(c.Err())
 			})
 		}
 	}

ifempty.go

@@ -21,7 +21,7 @@ func DefaultIfEmpty[T any](s ...T) Operator[T, T] {
 								select {
 								default:
 								case <-done:
-									o.Error(c.Cause())
+									o.Error(c.Err())
 									return
 								}
 

merge.go

@@ -54,7 +54,7 @@ func (ob mergeWithObservable[T]) Subscribe(c Context, o Observer[T]) {
 		select {
 		default:
 		case <-done:
-			o.Error(c.Cause())
+			o.Error(c.Err())
 			return
 		}
 	}
@@ -65,7 +65,7 @@ func (ob mergeWithObservable[T]) Subscribe(c Context, o Observer[T]) {
 		select {
 		default:
 		case <-done:
-			o.Error(c.Cause())
+			o.Error(c.Err())
 			return
 		}
 	}

multicast.go

@@ -133,7 +133,7 @@ func (m *multicast[T]) Subscribe(c Context, o Observer[T]) {
 			select {
 			default:
 			case <-done:
-				o.Error(c.Cause())
+				o.Error(c.Err())
 				return
 			}
 
@@ -157,7 +157,7 @@ func (m *multicast[T]) Subscribe(c Context, o Observer[T]) {
 			m.Mu.Lock()
 			m.Mobs.Delete(&o)
 			m.Mu.Unlock()
-			o.Error(c.Cause())
+			o.Error(c.Err())
 		})
 	}
 

observable.go

@@ -17,8 +17,8 @@ package rx
 //
 // An Observable must honor the cancellation of the given [Context].
 // When the cancellation of the given [Context] is detected, an Observable
-// must emit an error notification of whatever [Context.Cause] returns, as
-// a termination, to the given [Observer] as soon as possible.
+// must emit a notification of error (as a termination) to the given [Observer]
+// as soon as possible.
 //
 // If an Observable needs to start goroutines, it must use [Context.Go] to do
 // so; if an Observable needs to start an asynchronous operation other than

pair.go

@@ -27,7 +27,7 @@ func FromMap[M ~map[K]V, K comparable, V any](m M) Observable[Pair[K, V]] {
 			select {
 			default:
 			case <-done:
-				o.Error(c.Cause())
+				o.Error(c.Err())
 				return
 			}
 

range.go

@@ -12,7 +12,7 @@ func Range[T constraints.Integer](low, high T) Observable[T] {
 			select {
 			default:
 			case <-done:
-				o.Error(c.Cause())
+				o.Error(c.Err())
 				return
 			}
 
@@ -33,7 +33,7 @@ func Iota[T constraints.Integer](init T) Observable[T] {
 			select {
 			default:
 			case <-done:
-				o.Error(c.Cause())
+				o.Error(c.Err())
 				return
 			}
 

repeat.go

@@ -48,7 +48,7 @@ func (ob repeatObservable[T]) Subscribe(c Context, o Observer[T]) {
 		select {
 		default:
 		case <-done:
-			o.Error(c.Cause())
+			o.Error(c.Err())
 			return
 		}
 

retry.go

@@ -41,7 +41,7 @@ func (ob retryObservable[T]) Subscribe(c Context, o Observer[T]) {
 		select {
 		default:
 		case <-done:
-			o.Error(c.Cause())
+			o.Error(c.Err())
 			return
 		}
 

skipuntil.go

@@ -78,7 +78,7 @@ func (ob skipUntilObservable[T, U]) Subscribe(c Context, o Observer[T]) {
 	select {
 	default:
 	case <-c.Done():
-		terminate(Error[T](c.Cause()))
+		terminate(Error[T](c.Err()))
 		return
 	}
 

slice.go

@@ -10,7 +10,7 @@ func FromSlice[S ~[]T, T any](s S) Observable[T] {
 			select {
 			default:
 			case <-done:
-				o.Error(c.Cause())
+				o.Error(c.Err())
 				return
 			}
 

takelast.go

@@ -32,7 +32,7 @@ func TakeLast[T any](count int) Operator[T, T] {
 							select {
 							default:
 							case <-done:
-								o.Error(c.Cause())
+								o.Error(c.Err())
 								return
 							}
 

takeuntil.go

@@ -99,7 +99,7 @@ func (ob takeUntilObservable[T, U]) Subscribe(c Context, o Observer[T]) {
 	select {
 	default:
 	case <-c.Done():
-		terminate(Error[T](c.Cause()))
+		terminate(Error[T](c.Err()))
 		return
 	}
 

timing.go

@@ -17,7 +17,7 @@ func Ticker(d time.Duration) Observable[time.Time] {
 			for {
 				select {
 				case <-done:
-					o.Error(c.Cause())
+					o.Error(c.Err())
 					return
 				case t := <-tk.C:
 					Try1(o, Next(t), func() { o.Error(ErrOops) })
@@ -36,7 +36,7 @@ func Timer(d time.Duration) Observable[time.Time] {
 			select {
 			case <-c.Done():
 				tm.Stop()
-				o.Error(c.Cause())
+				o.Error(c.Err())
 			case t := <-tm.C:
 				Try1(o, Next(t), func() { o.Error(ErrOops) })
 				o.Complete()

unicast.go

@@ -96,7 +96,7 @@ func (u *unicast[T]) startEmitting(n Notification[T], unlockEarly bool) {
 		select {
 		default:
 		case <-u.DoneChan:
-			throw(context.Cause(u.Context))
+			throw(u.Context.Err())
 			return
 		}
 
@@ -149,7 +149,7 @@ func (u *unicast[T]) startEmitting(n Notification[T], unlockEarly bool) {
 			select {
 			default:
 			case <-u.DoneChan:
-				throw(context.Cause(u.Context))
+				throw(u.Context.Err())
 				return
 			}
 
@@ -245,7 +245,7 @@ func (u *unicast[T]) Subscribe(c Context, o Observer[T]) {
 
 	done := c.Done()
 	if done != nil {
-		stop := c.AfterFunc(func() { u.Emit(Error[T](c.Cause())) })
+		stop := c.AfterFunc(func() { u.Emit(Error[T](c.Err())) })
 		o = o.DoOnTermination(func() { stop() })
 	}