How to implement a configurable-interval 'keepalive' signal whenever a source is silent?

This answer is in response to your question and your answer.

You should try to avoid .Publish() when possible as it creates an IConnectableObservable<T> that can be a pain to manage. Where possible use the .Publish(inner => ...) overload that gives you a plain IObservable<T>.

In the main part of your query the Merge, Throttle and Never are all unnecessary.

You are right that you need a double Switch.

In your question and answer you didn't declare the types of observables - and it's a little unclear from your code. I'll assume that source is an IObservable<long> and that the final query is also IObservable<long> where -1 is sent after each interval when the source didn't send.

Here's the code:

IObservable<long> source = ...
IObservable<TimeSpan> heartbeatTimes = ...

IObservable<long> query =
    source
        .Do(x => SendItem(x))
        .StartWith(0)
        .Publish(pxs =>
            heartbeatTimes
                .Select(t =>
                    pxs
                        .StartWith(0)
                        .Select(px => Observable.Interval(t).Select(x => -1L).StartWith(px))
                        .Switch())
                .Switch());

Here's my test code:

void Main()
{
    var random = new Random();
    IObservable<long> source = Observable.Generate(0L, x => x < 100L, x => x + 1L, x => x, x => TimeSpan.FromSeconds(random.NextDouble() * 5.0));
    var heartbeatTimes = new Subject<TimeSpan>();
    
    IObservable<long> query = ...
    
    IDisposable subscription = query.Subscribe(x => SendHeartbeat());
    
    heartbeatTimes.OnNext(TimeSpan.FromSeconds(1.0));

    Thread.Sleep(TimeSpan.FromSeconds(10.0));
    heartbeatTimes.OnNext(TimeSpan.FromSeconds(0.1));
}

void SendHeartbeat()
{
    Console.Write("!");
}

void SendItem(long item)
{
    Console.Write($"{item}.");
}

My tests seemed to work as you wanted.

heartbeatTimes should be an observable that immediately produces a value otherwise the query won't return anything until it does.

OK, I'm quite proud of this, so if you're going to shoot it down, please do so gently :)

var items = source.Publish();
items.Subscribe(item => SendItem(item)); //not interesting, but for completeness' sake

heartbeatTimes.Select(time =>  //for each incoming heartbeat time
    //merge two sources, starts and stops, both derived from time:
    Observable.Merge(               
        //starts
        // (need initial value otherwise if no items, interval is never started)
        items.StartWith((0, null))     
            .Throttle(time)                        // when itemsource is silent for [time]
            .Select(_ => Observable.Interval(time) // start interval every [time]
                  // interval fires at end of [time], so send an initial trigger immediately
                 .StartWith(0)),                  
        //stops
        // as soon as a item comes, send Never (to replace Observable.Interval)
        items.Select(_ => Observable.Never<long>()) 
    ).Switch() //chose latest source from merged stops/starts
)
//switches from one start/stop source, to the next 
// (each time heartbeatTime produces a new value)
.Switch()  
.Subscribe(_ => SendHeartbeat());

items.Connect();