What causes "Value cannot be null. (Parameter 'source')" in dotnet/reactive?

Ensure the source is non-null before calling Aggregate: use the ?? operator to substitute Observable.Empty ().

dotnet/reactive · error · ArgumentNullException

Value cannot be null. (Parameter 'source')

Error message

Value cannot be null. (Parameter 'source')

What it means

The seeded Aggregate extension validates its arguments eagerly and throws ArgumentNullException when the source observable is null. This is a synchronous, immediate fail-fast check performed at call time so the caller gets a precise stack trace instead of a cryptic failure inside the pipeline.

Solutions

Ensure the source is non-null before calling Aggregate: use the ?? operator to substitute Observable.Empty<TSource>().
Fix the producer/method that returns null instead of an empty observable; returning null from an Rx factory is a bug.
Enable nullable reference types (C# 8 #nullable enable) so the compiler flags possibly-null sources.
Check arguments with Arg.NotNull / ArgumentNullException.ThrowIfNull at your own API boundary to catch it early.

Example fix

// before
var result = await GetStream().Aggregate(0, (acc, x) => acc + x); // NRE risk if GetStream() returns null

// after
var src = GetStream() ?? Observable.Empty<int>();
var result = await src.Aggregate(0, (acc, x) => acc + x);

Defensive patterns

Strategy: type-guard

Validate before calling

if (source is null) source = Observable.Empty<TSource>();
ArgumentNullException.ThrowIfNull(source); // optional explicit guard

Type guard

static IObservable<T> NotNull<T>(IObservable<T>? s) => s ?? Observable.Empty<T>();

Try / catch

try { var result = source.Aggregate(seed, acc); }
catch (ArgumentNullException ex) when (ex.ParamName == "source") { /* recover: substitute empty stream or report caller bug */ }

Prevention

Never let Rx factory methods return null; return Observable.Empty instead.
Enable C# nullable reference types so null sources are caught at compile time.
Coalesce nullable stream references with ?? Observable.Empty<T>() at call sites.
Validate inputs at your own API boundary before handing streams to Rx operators.

When it happens

Trigger: Invoking Observable.Aggregate(source, seed, accumulator) where the source IObservable<TSource> reference is null (e.g. a factory method returned null, an uninitialized field, or a dictionary lookup missed).

Common situations: Chaining operators off a method that can return null instead of an empty observable; conditional composition where the base stream was never assigned; DI containers yielding null for an unregistered stream dependency.

Related errors

AI-assisted analysis of dotnet/reactive@94b5d5ab91 (2026-09-15). Data as JSON: /api/errors/6e1ae06616258acd. Report an issue: GitHub.

Appendix: source

Thrown at Rx.NET/Source/src/System.Reactive/Linq/Observable.Aggregates.cs:30

        #region + Aggregate +

        /// <summary>
        /// Applies an accumulator function over an observable sequence, returning the result of the aggregation as a single element in the result sequence. The specified seed value is used as the initial accumulator value.
        /// For aggregation behavior with incremental intermediate results, see <see cref="Observable.Scan{TSource, Accumulate}"/>.
        /// </summary>
        /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
        /// <typeparam name="TAccumulate">The type of the result of the aggregation.</typeparam>
        /// <param name="source">An observable sequence to aggregate over.</param>
        /// <param name="seed">The initial accumulator value.</param>
        /// <param name="accumulator">An accumulator function to be invoked on each element.</param>
        /// <returns>An observable sequence containing a single element with the final accumulator value.</returns>
        /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="accumulator"/> is null.</exception>
        /// <remarks>The return type of this operator differs from the corresponding operator on IEnumerable in order to retain asynchronous behavior.</remarks>
        public static IObservable<TAccumulate> Aggregate<TSource, TAccumulate>(this IObservable<TSource> source, TAccumulate seed, Func<TAccumulate, TSource, TAccumulate> accumulator)
        {
            if (source == null)
            {
                throw new ArgumentNullException(nameof(source));
            }

            if (accumulator == null)
            {
                throw new ArgumentNullException(nameof(accumulator));
            }

            return s_impl.Aggregate(source, seed, accumulator);
        }

        /// <summary>
        /// Applies an accumulator function over an observable sequence, returning the result of the aggregation as a single element in the result sequence. The specified seed value is used as the initial accumulator value,
        /// and the specified result selector function is used to select the result value.
        /// </summary>
        /// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
        /// <typeparam name="TAccumulate">The type of the accumulator value.</typeparam>
        /// <typeparam name="TResult">The type of the resulting value.</typeparam>
        /// <param name="source">An observable sequence to aggregate over.</param>

View on GitHub (pinned to 94b5d5ab91)