fix(rx): fix average op gather next (#240)

snivilised · Apr 24, 2024 · 1fd4af2 · 1fd4af2
1 parent af00486
commit 1fd4af2
Showing 1 changed file with 4 additions and 8 deletions.
diff --git a/rx/observable-operator.go b/rx/observable-operator.go
@@ -156,14 +156,10 @@ func (op *averageOperator[T]) end(ctx context.Context, dst chan<- Item[T]) {
 func (op *averageOperator[T]) gatherNext(_ context.Context, item Item[T],
 	_ chan<- Item[T], _ operatorOptions[T],
 ) {
-	_ = item
-
-	// TODO(fix): v := item.V.(*averageOperator[T])
-	// op.calc.Add(op.sum)
-	// op.sum += v.sum
-	// op.count += v.count
-	//
-	panic("averageOperator.gatherNext NOT-IMPL")
+	if v, err := TryOpaque[T, *averageOperator[T]](item); err != nil {
+		op.sum = op.calc.Add(op.sum, v.sum)
+		op.count = op.calc.Add(op.count, v.count)
+	}
 }
 
 // BackOffRetry implements a backoff retry if a source Observable sends an error,