How to use ReactiveUI::WhenAnyValue correctly with alternate windows threads?

Question

I have built a small logging GUI to help debug the main application. It needs to run on it's own thread so it is not blocked when the main UI becomes unresponsive. Part of the purpose of the logging window is to find out why the main window becomes unresponsive.

The new window is launched via this method.

    public static Task<T> CreateAndShowStaWindow<T>(Func<T> factory) where T:Window
    {

        var windowResult = new TaskCompletionSource<T>();

        var newWindowThread = new Thread(() =>
        {
            var window = factory();
            window.Show();
            windowResult.SetResult( window );
            window.Events().Closed.Subscribe( _ => window.Dispatcher.InvokeShutdown() );
            System.Windows.Threading.Dispatcher.Run();

        });
        newWindowThread.SetApartmentState(ApartmentState.STA);
        newWindowThread.IsBackground = true;
        newWindowThread.Start();
        newWindowThread.Name = "STA WPF";

        return windowResult.Task;
    }

Inside the window I have a textbox

<TextBox x:Name="FilterText"/>

which I wish to monitor for changes using ReactiveUI WhenAnyValue

this
      .WhenAnyValue( p => p.FilterText.Text )
      .Subscribe( Console.WriteLine);

Now I am sure the above call to WhenAnyValue is run on the new STA thread. I can check this in the debugger.

and

If I let the program continue I get

InvalidOperationException: The calling thread cannot access this object because a different thread owns it.

This exception occurs on "MAIN" thread rather than the "STA WPF" thread and at this point in the stack

ReactiveUI.dll!ReactiveUI.Reflection.TryGetValueForPropertyChain<object>(out object changeValue, object current, System.Collections.Generic.IEnumerable<System.Linq.Expressions.Expression> expressionChain) Line 129   
ReactiveUI.dll!ReactiveUI.ReactiveNotifyPropertyChangedMixin.observedChangeFor(System.Linq.Expressions.Expression expression, ReactiveUI.IObservedChange<object, object> sourceChange) Line 134 
ReactiveUI.dll!ReactiveUI.ReactiveNotifyPropertyChangedMixin.nestedObservedChanges(System.Linq.Expressions.Expression expression, ReactiveUI.IObservedChange<object, object> sourceChange, bool beforeChange) Line 142
ReactiveUI.dll!ReactiveUI.ReactiveNotifyPropertyChangedMixin.SubscribeToExpressionChain.AnonymousMethod__1(ReactiveUI.IObservedChange<object,> object> y) Line 104  

With WPF it is not allowed to make cross thread access to WPF dependency properties. The question is why is ReactiveUI making a cross thread access?

Note that the error is not happening in the Console.WriteLine callback as passed to the subscription. It occurs inside the reactiveui code as it tries to read the Text property of the TextBox. Adding

this
      .WhenAnyValue( p => p.FilterText.Text )
      .ObserveOn(DispatcherScheduler.Current)
      .Subscribe( Console.WriteLine);

does not fix the problem. Neither does

this
     .WhenAnyValue(p => p.FilterText.Text)
     .SubscribeOn( DispatcherScheduler.Current )
     .Subscribe(Console.WriteLine);

As a santiy check if I remove the WhenAnyValue call from the code then I don't get this error.

It seems that ReactiveUI is getting confused with which synchronization context it should be operating on. But maybe I'm doing something wrong. Is there a solution / workaround to this problem?

EDIT

To reproduce the error you will need to insert a call to

   Locator.CurrentMutable.InitializeReactiveUI();

some point early in the application startup. A project that replicates the bug is here.

https://github.com/bradphelan/RxUIBug1375

Answer 1

But maybe I'm doing something wrong

Probably because it works for me if I create the window like this:

Thread newWindowThread = new Thread(new ThreadStart(() =>
{
    SynchronizationContext.SetSynchronizationContext(
        new DispatcherSynchronizationContext(
            Dispatcher.CurrentDispatcher));

    RxWindow tempWindow = new RxWindow();
    tempWindow.Closed += (ss, ee) => Dispatcher.CurrentDispatcher.BeginInvokeShutdown(DispatcherPriority.Background);
    tempWindow.Show();

    Dispatcher.Run();
}));
newWindowThread.SetApartmentState(ApartmentState.STA);
newWindowThread.IsBackground = true;
newWindowThread.Start();
newWindowThread.Name = "STA WPF";

RxWindow.xaml.cs:

public partial class RxWindow : Window
{
    public RxWindow()
    {
        InitializeComponent();

        this.WhenAnyValue(p => p.FilterText.Text)
            .Subscribe(_ => MessageBox.Show(""));
    }
}

Answer 2

[Update]

I didn't understand the question correctly. You could probably use SubscribeOn, but having it default to the calling context instead of the main thread would be nicer indeed.

[Old]

Your breakpoint is set on the line defining the WhenAny call. When the WhenAny subscription is defined, you can be sure it's on your STA thread. But the subscription can (and often will) observe on a new thread because that's how Rx works; asynchronous.

You can confirm this by putting a breakpoint on the method you pass to Subscribe(), you'll see it's being called from a different thread.

You can use the .ObserveOn() method to define on which thread the subscription will be observed.

RxUI has an IScheduler set up to reference your main thread: RxApp.MainThreadScheduler, you can pass this directly to .ObserveOn().

this
  .WhenAnyValue( p => p.FilterText.Text )
  .ObserveOn( RxApp.MainThreadScheduler )
  .Subscribe( Console.WriteLine);