What is ObserveOnTimeProvider intended / useful for?

[JaggerJo]

Really interesting project. We're heavily using RxNet and could benefit from a lower overhead implementation.

I was looking through the sources and especially into how ObserveOn(..) is implemented. After looking into it I don't get what ObserveOnTimeProvider is for. An explanation would be appreciated.

ObserveOnTimeProvider is only created when ObserveOn is called with the non system TimeProvider.

The internal _Observer class of the ObserveOnTimeProvider creates a stopped timer that (It seems) never elapses.

R3/src/R3/Internal/TimeProviderExtensions.cs

Lines 5 to 8 in 9b18209

	public static ITimer CreateStoppedTimer(this TimeProvider timeProvider, TimerCallback timerCallback, object? state)
	{
	return timeProvider.CreateTimer(timerCallback, state, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
	}

After values arrive in the _Observer this extension method is called on the timer:

R3/src/R3/Internal/TimeProviderExtensions.cs

Lines 10 to 13 in 9b18209

	public static void RestartImmediately(this ITimer timer)
	{
	timer.Change(TimeSpan.Zero, Timeout.InfiniteTimeSpan);
	}

According to the docs this timer is invoked immediately but never called periodically.

In this test case the FakeTimeProvider is not advanced, but the elements still end up in the output list.

R3/tests/R3.Tests/OperatorTests/ObserveOnTest.cs

Lines 34 to 50 in 9b18209

	public void TimeProvider()
	{
	var fakeTime = new FakeTimeProvider();
	var publisher = new Subject<int>();
	using var list = publisher.ObserveOn(fakeTime).ToLiveList();
	publisher.OnNext(10);
	publisher.OnNext(20);
	publisher.OnNext(30);
	list.AssertEqual([10, 20, 30]);
	publisher.OnCompleted();
	list.AssertIsCompleted();
	}

What is the ObserveOnTimeProvider used for if it just immediately passes on messages, even if time stands still from the timers perspective?

[neuecc]

TimeProvider serves the role of IScheduler in RxNet.
In other words, ObserverOnTimeProvider is equivalent to ObserveOnScheduler.
For example, if you pass WpfDispatcherTimerProvider (which is the TimeProvider implementation for WPF provided by R3), you can place values on the Dispatcher (UI thread).

[JaggerJo]

@neuecc Thanks for the quick reply.

Might be a silly question, but why do you need a timer for it if you only ever use the run this callback ... portion? I really want to get it. I think I can learn something here :)

https://learn.microsoft.com/en-us/dotnet/api/system.timeprovider.createtimer?view=net-8.0#parameters

A delegate representing a method to be executed when the timer fires. The method specified for callback should be reentrant, as it may be invoked simultaneously on two threads if the timer fires again before or while a previous callback is still being handled.

[neuecc]

First, generating and executing a Timer for each individual value must absolutely be avoided for performance reasons.
Therefore, it uses a single Timer and maintains a local Queue for when new values arrive during delegate execution.
ObserveOn is a feature that emits values in the execution context provided by the TimeProvider.
Thus, with SystemTimeProvider it's the ThreadPool, and with WpfDispatcherTimeProvider it's the Dispatcher.
In the case of FakeTimeProvider, since it doesn't have an execution context, it becomes Immediate.
Since this isn't about waiting for a specific time period, it's set to fire just once immediately.
By using Timer's Change method, we can execute the delegate once in the execution context each time without having to generate new Timers.
In RxNet terms, we're simulating something like IScheduler.Schedule(delegate) using Timer.

By the way, giving special treatment to TimeProvider.System by making it ObserveOnThreadPool is done for higher performance optimization.
Even if you were to CreateTimer with SystemTimeProvider, the behavior itself would be the same.