update(*): rewrite *Complete* -> *Success*

audit.go

@@ -41,9 +41,9 @@ func (obs auditObservable[T, U]) Subscribe(c Context, sink Observer[T]) {
 	sink = sink.OnLastNotification(cancel)
 
 	var x struct {
-		Context  atomic.Value
-		Complete atomic.Bool
-		Latest   struct {
+		Context atomic.Value
+		Success atomic.Bool
+		Latest  struct {
 			sync.Mutex
 			Value T
 		}
@@ -85,18 +85,18 @@ func (obs auditObservable[T, U]) Subscribe(c Context, sink Observer[T]) {
 					}
 				})
 
-				if x.Context.CompareAndSwap(w.Context, c.Context) && x.Complete.Load() && x.Context.CompareAndSwap(c.Context, sentinel) {
-					sink.Complete()
+				if x.Context.CompareAndSwap(w.Context, c.Context) && x.Success.Load() && x.Context.CompareAndSwap(c.Context, sentinel) {
+					sink.Success()
 				}
 
 			case KindError:
 				if x.Context.Swap(sentinel) != sentinel {
 					sink.Error(n.Error)
 				}
 
-			case KindComplete:
-				if x.Context.CompareAndSwap(w.Context, c.Context) && x.Complete.Load() && x.Context.CompareAndSwap(c.Context, sentinel) {
-					sink.Complete()
+			case KindSuccess:
+				if x.Context.CompareAndSwap(w.Context, c.Context) && x.Success.Load() && x.Context.CompareAndSwap(c.Context, sentinel) {
+					sink.Success()
 				}
 			}
 		})
@@ -123,8 +123,8 @@ func (obs auditObservable[T, U]) Subscribe(c Context, sink Observer[T]) {
 				sink(n)
 			}
 
-		case KindComplete:
-			x.Complete.Store(true)
+		case KindSuccess:
+			x.Success.Store(true)
 
 			if x.Context.CompareAndSwap(c.Context, sentinel) {
 				sink(n)

audit_test.go

@@ -21,7 +21,7 @@ func TestAudit(t *testing.T) {
 				},
 			),
 		),
-		"B", "D", "E", ErrComplete,
+		"B", "D", "E", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
@@ -32,7 +32,7 @@ func TestAudit(t *testing.T) {
 				},
 			),
 		),
-		ErrComplete,
+		ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
@@ -46,7 +46,7 @@ func TestAudit(t *testing.T) {
 				},
 			),
 		),
-		ErrComplete,
+		ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
@@ -77,13 +77,13 @@ func TestAudit(t *testing.T) {
 			AddLatencyToValues[string](1, 2),
 			rx.AuditTime[string](Step(3)),
 		),
-		"B", "D", "E", ErrComplete,
+		"B", "D", "E", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Empty[string](),
 			rx.AuditTime[string](Step(3)),
 		),
-		ErrComplete,
+		ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Throw[string](ErrTest),

blocking.go

@@ -19,7 +19,7 @@ func (obs Observable[T]) BlockingFirst(parent Context) (v T, err error) {
 		}
 
 		switch n.Kind {
-		case KindNext, KindError, KindComplete:
+		case KindNext, KindError, KindSuccess:
 			noop = true
 
 			switch n.Kind {
@@ -81,7 +81,7 @@ func (obs Observable[T]) BlockingLast(parent Context) (v T, err error) {
 		}
 
 		switch n.Kind {
-		case KindError, KindComplete:
+		case KindError, KindSuccess:
 			cancel()
 		}
 	})
@@ -149,7 +149,7 @@ func (obs Observable[T]) BlockingSingle(parent Context) (v T, err error) {
 		}
 
 		switch n.Kind {
-		case KindError, KindComplete:
+		case KindError, KindSuccess:
 			cancel()
 		}
 	})
@@ -187,7 +187,7 @@ func (obs Observable[T]) BlockingSubscribe(parent Context, sink Observer[T]) err
 		res = n
 		sink(n)
 		switch n.Kind {
-		case KindError, KindComplete:
+		case KindError, KindSuccess:
 			cancel()
 		}
 	})
@@ -203,7 +203,7 @@ func (obs Observable[T]) BlockingSubscribe(parent Context, sink Observer[T]) err
 	switch res.Kind {
 	case KindError:
 		return res.Error
-	case KindComplete:
+	case KindSuccess:
 		return nil
 	default:
 		panic("unreachable")

buffercount.go

@@ -80,7 +80,7 @@ func (obs bufferCountObservable[T]) Subscribe(c Context, sink Observer[[]T]) {
 		case KindError:
 			sink.Error(n.Error)
 
-		case KindComplete:
+		case KindSuccess:
 			if len(s) != 0 {
 				for {
 					Try1(sink, Next(s), func() { sink.Error(ErrOops) })
@@ -93,7 +93,7 @@ func (obs bufferCountObservable[T]) Subscribe(c Context, sink Observer[[]T]) {
 				}
 			}
 
-			sink.Complete()
+			sink.Success()
 		}
 	})
 }

buffercount_test.go

@@ -16,35 +16,35 @@ func TestBufferCount(t *testing.T) {
 			rx.BufferCount[string](2),
 			ToString[[]string](),
 		),
-		"[A B]", "[C D]", "[E]", ErrComplete,
+		"[A B]", "[C D]", "[E]", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
 			rx.BufferCount[string](3),
 			ToString[[]string](),
 		),
-		"[A B C]", "[D E]", ErrComplete,
+		"[A B C]", "[D E]", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
 			rx.BufferCount[string](3).WithStartBufferEvery(1),
 			ToString[[]string](),
 		),
-		"[A B C]", "[B C D]", "[C D E]", "[D E]", "[E]", ErrComplete,
+		"[A B C]", "[B C D]", "[C D E]", "[D E]", "[E]", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
 			rx.BufferCount[string](3).WithStartBufferEvery(2),
 			ToString[[]string](),
 		),
-		"[A B C]", "[C D E]", "[E]", ErrComplete,
+		"[A B C]", "[C D E]", "[E]", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Just("A", "B", "C", "D", "E"),
 			rx.BufferCount[string](3).WithStartBufferEvery(4),
 			ToString[[]string](),
 		),
-		"[A B C]", "[E]", ErrComplete,
+		"[A B C]", "[E]", ErrSuccess,
 	).Case(
 		rx.Pipe2(
 			rx.Throw[string](ErrTest),

catch.go

@@ -1,7 +1,7 @@
 package rx
 
-// Catch handles errors on the source Observable by mirroring a new Observable
-// returned by selector.
+// Catch mirrors the source or switches to another Observable, returned from
+// a call to selector, instead if the source emits an error notification.
 //
 // Catch does not catch context cancellations.
 func Catch[T any](selector func(err error) Observable[T]) Operator[T, T] {
@@ -24,7 +24,7 @@ func Catch[T any](selector func(err error) Observable[T]) Operator[T, T] {
 						obs := Try11(selector, n.Error, func() { sink.Error(ErrOops) })
 						obs.Subscribe(c, sink)
 
-					case KindComplete:
+					case KindSuccess:
 						sink(n)
 					}
 				})
@@ -56,7 +56,7 @@ func OnErrorResumeWith[T any](obs Observable[T]) Operator[T, T] {
 
 						obs.Subscribe(c, sink)
 
-					case KindComplete:
+					case KindSuccess:
 						sink(n)
 					}
 				})
@@ -65,11 +65,11 @@ func OnErrorResumeWith[T any](obs Observable[T]) Operator[T, T] {
 	)
 }
 
-// OnErrorComplete mirrors the source Observable, or completes if the source
-// emits an error notification.
+// OnErrorSucceed mirrors the source Observable, or succeeds instead
+// if the source emits an error notification.
 //
-// OnErrorComplete does not complete after context cancellation.
-func OnErrorComplete[T any]() Operator[T, T] {
+// OnErrorSucceed does not succeed after context cancellation.
+func OnErrorSucceed[T any]() Operator[T, T] {
 	return NewOperator(
 		func(source Observable[T]) Observable[T] {
 			return func(c Context, sink Observer[T]) {
@@ -86,9 +86,9 @@ func OnErrorComplete[T any]() Operator[T, T] {
 							return
 						}
 
-						sink.Complete()
+						sink.Success()
 
-					case KindComplete:
+					case KindSuccess:
 						sink(n)
 					}
 				})

catch_test.go

@@ -20,7 +20,7 @@ func TestCatch(t *testing.T) {
 			rx.Just("A", "B", "C"),
 			rx.Catch(f),
 		),
-		"A", "B", "C", ErrComplete,
+		"A", "B", "C", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Concat(
@@ -29,7 +29,7 @@ func TestCatch(t *testing.T) {
 			),
 			rx.Catch(f),
 		),
-		"A", "B", "C", "D", "E", ErrComplete,
+		"A", "B", "C", "D", "E", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Throw[string](ErrTest),
@@ -58,7 +58,7 @@ func TestOnErrorResumeWith(t *testing.T) {
 			rx.Just("A", "B", "C"),
 			rx.OnErrorResumeWith(rx.Just("D", "E")),
 		),
-		"A", "B", "C", ErrComplete,
+		"A", "B", "C", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Concat(
@@ -67,7 +67,7 @@ func TestOnErrorResumeWith(t *testing.T) {
 			),
 			rx.OnErrorResumeWith(rx.Just("D", "E")),
 		),
-		"A", "B", "C", "D", "E", ErrComplete,
+		"A", "B", "C", "D", "E", ErrSuccess,
 	)
 
 	ctx, cancel := rx.NewBackgroundContext().WithTimeout(Step(1))
@@ -82,24 +82,24 @@ func TestOnErrorResumeWith(t *testing.T) {
 	)
 }
 
-func TestOnErrorComplete(t *testing.T) {
+func TestOnErrorSucceed(t *testing.T) {
 	t.Parallel()
 
 	NewTestSuite[string](t).Case(
 		rx.Pipe1(
 			rx.Just("A", "B", "C"),
-			rx.OnErrorComplete[string](),
+			rx.OnErrorSucceed[string](),
 		),
-		"A", "B", "C", ErrComplete,
+		"A", "B", "C", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Concat(
 				rx.Just("A", "B", "C"),
 				rx.Throw[string](ErrTest),
 			),
-			rx.OnErrorComplete[string](),
+			rx.OnErrorSucceed[string](),
 		),
-		"A", "B", "C", ErrComplete,
+		"A", "B", "C", ErrSuccess,
 	)
 
 	ctx, cancel := rx.NewBackgroundContext().WithTimeout(Step(1))
@@ -108,7 +108,7 @@ func TestOnErrorComplete(t *testing.T) {
 	NewTestSuite[string](t).WithContext(ctx).Case(
 		rx.Pipe1(
 			rx.Never[string](),
-			rx.OnErrorComplete[string](),
+			rx.OnErrorSucceed[string](),
 		),
 		context.DeadlineExceeded,
 	)

channelize.go

@@ -33,7 +33,7 @@ func Channelize[T any](join func(upstream <-chan Notification[T], downstream cha
 								c.Go(func() { drain(downstream) })
 								sink.Error(ErrOops)
 							})
-						case KindError, KindComplete:
+						case KindError, KindSuccess:
 							defer drain(downstream)
 							sink(n)
 							return

channelize_test.go

@@ -15,7 +15,7 @@ func TestChannelize(t *testing.T) {
 			switch n.Kind {
 			case rx.KindNext:
 				downstream <- n
-			case rx.KindError, rx.KindComplete:
+			case rx.KindError, rx.KindSuccess:
 				downstream <- n
 				return
 			}
@@ -27,7 +27,7 @@ func TestChannelize(t *testing.T) {
 			rx.Just("A", "B", "C"),
 			rx.Channelize(join),
 		),
-		"A", "B", "C", ErrComplete,
+		"A", "B", "C", ErrSuccess,
 	).Case(
 		rx.Pipe1(
 			rx.Just("A", "B", "C"),

combinelatest2.go

@@ -70,9 +70,9 @@ func combineLatestTry2[T1, T2, R, X any](
 		sink.Error(n.Error)
 		return false
 
-	case KindComplete:
+	case KindSuccess:
 		if s.CBits |= bit; s.CBits == FullBits {
-			sink.Complete()
+			sink.Success()
 			return false
 		}
 	}

combinelatest2_test.go

@@ -24,7 +24,7 @@ func TestCombineLatest2(t *testing.T) {
 		"[A1 B1]",
 		"[A2 B1]",
 		"[A2 B2]",
-		ErrComplete,
+		ErrSuccess,
 	).Case(
 		rx.CombineLatest2(
 			rx.Throw[string](ErrTest),

combinelatest3.go

@@ -76,9 +76,9 @@ func combineLatestTry3[T1, T2, T3, R, X any](
 		sink.Error(n.Error)
 		return false
 
-	case KindComplete:
+	case KindSuccess:
 		if s.CBits |= bit; s.CBits == FullBits {
-			sink.Complete()
+			sink.Success()
 			return false
 		}
 	}

combinelatest3_test.go

@@ -26,7 +26,7 @@ func TestCombineLatest3(t *testing.T) {
 		"[A2 B1 C1]",
 		"[A2 B2 C1]",
 		"[A2 B2 C2]",
-		ErrComplete,
+		ErrSuccess,
 	).Case(
 		rx.CombineLatest3(
 			rx.Throw[string](ErrTest),