[BUG] Sample code from Handbook is not working

[SirRufo]

Describe the bug

The given sample from Handbook/MessageBus/Ways to avoid using MessageBus is not working. Only the last added item is handled correct (removed from the list).

It worked fine on version 10.5.43 and failed after that (>=11.*).

Steps To Reproduce

using DynamicData;
using DynamicData.Binding;

using ReactiveUI;

using System;
using System.Collections.ObjectModel;
using System.Linq;
using System.Reactive;
using System.Reactive.Linq;

using Xunit;

namespace ReactiveUITests
{
    /// <summary>
    /// <code>https://www.reactiveui.net/docs/handbook/message-bus/#ways-to-avoid-using-messagebus</code>
    /// </summary>
    public class MessageBusWaysToAvoidUsingMessageBusTest
    {
        public class DocumentViewModel : ReactiveObject
        {
            public ReactiveCommand<Unit, Unit> Close { get; set; }

            public DocumentViewModel()
            {
                // Note that we don't actually *subscribe* to Close here or implement
                // anything in DocumentViewModel, because Closing is a responsibility
                // of the document list.
                Close = ReactiveCommand.Create(() => { });
            }
        }

        public class MainViewModel : ReactiveObject
        {
            public ObservableCollection<DocumentViewModel> OpenDocuments { get; protected set; }

            public MainViewModel()
            {
                OpenDocuments = new ObservableCollection<DocumentViewModel>();

                // Whenever the list of documents change, calculate a new Observable
                // to represent whenever any of the *current* documents have been
                // requested to close, then Switch to that. When we get something
                // to close, remove it from the list.
                OpenDocuments
                    .ToObservableChangeSet()
                    .AutoRefreshOnObservable(document => document.Close)
                    .Select(_ => WhenAnyDocumentClosed())
                    .Switch()
                    .Subscribe(x => OpenDocuments.Remove(x));
            }

            IObservable<DocumentViewModel> WhenAnyDocumentClosed()
            {
                // Select the documents into a list of Observables
                // who return the Document to close when signaled,
                // then flatten them all together.
                return OpenDocuments
                    .Select(x => x.Close.Select(_ => x))
                    .Merge();
            }
        }

        [Fact]
        public void TestWithSingleDocument()
        {
            var main = new MainViewModel();
            var doc = new DocumentViewModel();
            main.OpenDocuments.Add(doc);

            doc.Close.Execute().Subscribe();
            var actual = main.OpenDocuments.Count;

            Assert.Equal(0, actual);
        }

        [Fact]
        public void TestWithFirstFromMultipleDocuments()
        {
            var main = new MainViewModel();
            var doc = new DocumentViewModel();
            main.OpenDocuments.Add(doc);
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());

            doc.Close.Execute().Subscribe();
            var actual = main.OpenDocuments.Count;

            Assert.Equal(4, actual);
        }

        [Fact]
        public void TestWithCenterFromMultipleDocuments()
        {
            var main = new MainViewModel();
            var doc = new DocumentViewModel();
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(doc);
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());

            doc.Close.Execute().Subscribe();
            var actual = main.OpenDocuments.Count;

            Assert.Equal(4, actual);
        }

        [Fact]
        public void TestWithLastFromMultipleDocuments()
        {
            var main = new MainViewModel();
            var doc = new DocumentViewModel();
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(new DocumentViewModel());
            main.OpenDocuments.Add(doc);

            doc.Close.Execute().Subscribe();
            var actual = main.OpenDocuments.Count;

            Assert.Equal(4, actual);
        }
    }
}

Environment

• OS: Windows
• Version 10.0.19044.1466
• ReactiveUI Version: 11.* -> 17.1.17

[tomasfil]

It seems to be some kind of parallelism issue, because this fixes it:

                OpenDocuments
                    .ToObservableChangeSet()
                    .AutoRefreshOnObservable(document => document.Close)
//This helps
                    .ObserveOn(RxApp.MainThreadScheduler)
                    .Select(_ => WhenAnyDocumentClosed())
                    .Switch()
                    .Subscribe(x => OpenDocuments.Remove(x));

I am giving deeper look into this