What causes "observer" in dotnet/reactive?

Pass a non-null IObserver to Subscribe

dotnet/reactive · error · ArgumentNullException

observer

Error message

observer

What it means

Observable.ConcatMany's Subscribe validates its observer argument and throws ArgumentNullException when a null IObserver<T> is passed. This is standard Rx argument validation performed eagerly at subscription time rather than inside the observer pipeline.

Solutions

Pass a non-null IObserver<T> to Subscribe
Check the observer for null before calling Subscribe
Use Observer.Create<T>(...) to build a valid observer instead of passing null

Example fix

// before
observable.Subscribe(null);
// after
var observer = Observer.Create<int>(x => Console.WriteLine(x));
observable.Subscribe(observer);

Defensive patterns

Strategy: type-guard

Validate before calling

if (observer == null) throw new ArgumentNullException(nameof(observer));

Type guard

bool IsValidObserver<T>(IObserver<T> o) => o is not null;

Try / catch

try { observable.Subscribe(observer); }
catch (ArgumentNullException ex) { Log(ex); }

Prevention

Never pass a possibly-null observer; build one with Observer.Create
Validate factory results before subscribing
Enable nullable reference types to catch nulls at compile time

When it happens

Trigger: Calling Subscribe(null) directly on the ConcatMany observable, or a custom SubscribeOn/observer wrapper forwarding a null observer.

Common situations: Hand-rolled Rx pipelines where the observer is constructed dynamically and can be null; passing the result of a factory method that returned null.

Related errors

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

Appendix: source

Thrown at Rx.NET/Source/src/System.Reactive/Linq/Observable/ConcatMany.cs:24

using System.Reactive.Disposables;
using System.Threading;

namespace System.Reactive.Linq.ObservableImpl
{
    internal sealed class ConcatMany<T> : IObservable<T>
    {
        private readonly IObservable<IObservable<T>> _sources;

        internal ConcatMany(IObservable<IObservable<T>> sources)
        {
            _sources = sources;
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            if (observer == null)
            {
                throw new ArgumentNullException(nameof(observer));
            }

            var parent = new ConcatManyOuterObserver(observer);

            var d = _sources.SubscribeSafe(parent);
            parent.OnSubscribe(d);

            return parent;
        }

        internal sealed class ConcatManyOuterObserver : IObserver<IObservable<T>>, IDisposable
        {
            private readonly IObserver<T> _downstream;
            private readonly ConcurrentQueue<IObservable<T>> _queue;
            private readonly InnerObserver _innerObserver;

            private SingleAssignmentDisposableValue _upstream;
            private int _trampoline;

View on GitHub (pinned to 94b5d5ab91)