Valid setup for Reactive data polling in C#

I need to implement periodic polling of some data from remote component.
It should poll data next time after previous polling is done + wait let's say 5sec.
I came up with the code below but later figured out app throws Exception upon close - Event viewer has a record for 0xc00000fd which is stack overflow exception.
It's worth to mention that this exception only happens when app was open and polling data for quite a while (it takes time for stack to overflow).
All this is a WPF app and code below is in ViewModel.

I realize why exception is happening in this code (probably shouldn't call OnNext in Subscribe) but what is the correct way to implement?

_ctrlSubj = new Subject<ControllerInfo>();
_ctrlSubj.SelectMany(async _ => 
{
    // CurrentController is of type ControllerInfo
    // next line can take various amount of time
    var jobDetails = await Library.GetJobsAsync(CurrentController);
    return jobDetails;
})
.ObserveOnDispatcher()
.Subscribe(async e =>
{
    // Jobs is bound to View
    Jobs = new ObservableCollection<JobDetail>(jobDetails);

    await Task.Delay(TimeSpan.FromSeconds(5));
    _ctrlSubj.OnNext(CurrentController);
});

2nd edit:

@Aron's answer reminded me of multi subscription problems, which the below answer has. I recommend a helper function IntervalAsync looking like this:

public static class RxExtensions
{
    public static IObservable<TResult> IntervalAsync<TResult>(Func<IObservable<TResult>> f, TimeSpan period)
    {
        return IntervalAsync(f, period, Scheduler.Default);
    }

    public static IObservable<TResult> IntervalAsync<TResult>(Func<IObservable<TResult>> f, TimeSpan period, IScheduler scheduler)
    {
        return Observable.Create<TResult>(o =>
        {
            var q = new BehaviorSubject<TimeSpan>(TimeSpan.Zero);
            var observable = q
                .Delay(t => Observable.Timer(t))
                .SelectMany(_ => f())
                .Do(t => q.OnNext(period));
            return observable.Subscribe(o);
        });
    }
}

with the final code looking like this:

var subscription = RxExtensions.IntervalAsync(
        () => Observable.FromAsync(() => Library.GetJobsAsync(CurrentController)), 
        TimeSpan.FromSeconds(5)
    )
    .ObserveOnDispatcher()
    .Subscribe(i =>
    {
        Jobs = new ObservableCollection<JobDetail>(jobDetails);
    });

@Aron's answer works. I don't find it simpler because you have more Rx-TPL mixing, though I'l admit that 'simpler' is in the eye of the beholder.

First edit: (Not recommended, multiple subscription bugs are present).

Your concern is valid: With a synchronous reactive pipeline, Interval would wait for the full pipeline to finish. But with an async pipeline, Interval won't hold up. So if the first async task took 4.5 seconds, the next task would start .5 seconds after the first finished.

If you want the end-to-begin delay to be a set timespan, I think it would be best to do a queueing mechanism, similar to what you set up. I would do something similar to this:

var q = new BehaviorSubject<TimeSpan>(TimeSpan.Zero);
var subscription = q
    .Delay(t => Observable.Timer(t))
    .SelectMany(_ => Observable.FromAsync(() => Library.GetJobsAsync(CurrentController))
    .Subscribe(i =>
    {
        Jobs = new ObservableCollection<JobDetail>(jobDetails);
    });

Rx's thread & stack management I think works better here than TPL, and this won't lead to an infinite stack. However, I haven't tested it.

Original answer:

This may do it, but I can't test because no types.

var subscription = Observable.Interval(TimeSpan.FromSeconds(5))
    .SelectMany(_ => Observable.FromAsync(() => Library.GetJobsAsync(CurrentController))
    .ObserveOnDispatcher()
    .Subscribe(jobDetails =>
    {
        Jobs = new ObservableCollection<JobDetail>(jobDetails);
    });

If no good, then please modify your answer to include a mcve.

Testing code:

This is the code I'm using to test for async-TPL/RX mix. Doesn't full replicate @IgorStack's environment because no WPF (no .ObserveOnDispatcher()):

var f = new Func<Task<int>>(async () => {
    await Task.Delay(TimeSpan.FromSeconds(1));
    return 3;
});

var scheduler = new EventLoopScheduler(); //or Scheduler.Default
var o1 = RxExtensions.IntervalAsync(() => Observable.FromAsync(() => f()), TimeSpan.FromSeconds(5), scheduler)
    .Timestamp();
var subscription1 = o1.Subscribe(i =>
    {
        Console.WriteLine("s1: " + i.Timestamp.ToString("hh:mm:ss.ffff"));
    });
var subscription2 = o1.Subscribe(i =>
    {
        Console.WriteLine("s2: " + i.Timestamp.ToString("hh:mm:ss.ffff"));
    });