Rx DelayOnTrue extension implementation

It's an interesting issue and I do wonder whether it might be an XY kind of problem. Regardless, given your marble diagram you almost certainly don't want a DistinctUntilChanged() otherwise you'll never get repeating "F"s.

My approach to this (and there are undoubtedly others) would be to schedule the desired output ensuring to remove any values that get "throttled out" (i.e. a T followed by an F within the delay period). This can be achieved like this:

public static IObservable<bool> ThrottleOnTrue(this IObservable<bool> source, TimeSpan delay, IScheduler scheduler)
{
    return Observable.Create<bool>(
        observer =>
        {
            var serialDisposable = new SerialDisposable();

            var delays = source
                .Materialize()
                .Scan(
                    (Notification: (Notification<bool>)null, Delay: 0L),
                    (seed, source) => source.Kind switch
                        {
                            NotificationKind.OnCompleted => (Notification.CreateOnCompleted<bool>(), seed.Delay),
                            NotificationKind.OnError => (Notification.CreateOnError<bool>(source.Exception), 0),
                            _ => source.Value
                                ? (Notification.CreateOnNext(source.Value), delay.Ticks + 1)
                                : (Notification.CreateOnNext(source.Value), 1)
                        })
                .Where(tuple => tuple.Notification != null)
                .Publish();

            // Emit values after the delay, cancelling an items that are throttled
            var onNext = delays
                .Where(tuple => tuple.Notification.Kind == NotificationKind.OnNext)
                .Subscribe(tuple => serialDisposable.Disposable = scheduler.Schedule(scheduler.Now.AddTicks(tuple.Delay), () => observer.OnNext(tuple.Notification.Value)));

            // Emit completion after delay of last item to be emitted
            var onCompleted = delays
                .Where(tuple => tuple.Notification.Kind == NotificationKind.OnCompleted)
                .Subscribe(tuple => scheduler.Schedule(scheduler.Now.AddTicks(tuple.Delay), () => observer.OnCompleted()));

            // Emit errors immediately, cancelling any pending items
            var onError = delays
                .Where(tuple => tuple.Notification.Kind == NotificationKind.OnError)
                .Subscribe(tuple => serialDisposable.Disposable = scheduler.Schedule(TimeSpan.Zero, () => observer.OnError(tuple.Notification.Exception)));

            return new CompositeDisposable(new IDisposable[] { onNext, onCompleted, onError, delays.Connect(), serialDisposable });
        }
    );
}

It looks a little complicated due to the need to handle completion after the previously delayed item (which we keep track of with the Scan tuple).

Anyway, notice the addition of the scheduler parameter. An IScheduler parameter should always be provided when adding any form of asynchrony to Rx but can be defaulted to Scheduler.Default as shown here:

public static IObservable<bool> ThrottleOnTrue(this IObservable<bool> source, TimeSpan delay)
{
    return source.ThrottleOnTrue(delay, Scheduler.Default);
}

Now, the ThrottleOnTrue can be shown to work (with slightly different marble diagrams) by using "virtual time" courtesy of the TestScheduler. Here's a test showing that "F" values are emitted immediately (current time + 1 tick for scheduling):

private static long SchedulerOffset = ReactiveTest.Created + ReactiveTest.Subscribed;
private static long NotificationOffset = ReactiveTest.Subscribed;

/// <summary>
/// source:   F---F---F-C
/// expected: -F---F---F-C
/// </summary>
[Test]
public void ShouldDirectlyPassFalseValues()
{
    var scheduler = new TestScheduler();

    var source = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(2).Ticks, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + 1, Notification.CreateOnCompleted<bool>())
    };

    var expected = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(1).Ticks + NotificationOffset + 1, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(2).Ticks + NotificationOffset + 1, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + NotificationOffset + 1, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + NotificationOffset + 2, Notification.CreateOnCompleted<bool>())
    };

    var xs = scheduler
        .CreateColdObservable(source)
        .ThrottleOnTrue(TimeSpan.FromMinutes(1), scheduler);

    var observed = scheduler.Start(() => xs, TimeSpan.FromSeconds(3).Ticks + SchedulerOffset + 2);

    CollectionAssert.AreEqual(expected, observed.Messages);
}

And here's a test that showing that "T" values are emitted after the expected delay (and with previous "T" values cancelled):

/// <summary>
/// source:   T---T---T-C
/// expected: --------{delay}-T-C
/// </summary>
[Test]
public void ShouldDelayAndThrottleTrueValues()
{
    var scheduler = new TestScheduler();

    var source = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext(true)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(2).Ticks, Notification.CreateOnNext(true)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks, Notification.CreateOnNext(true)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + 1, Notification.CreateOnCompleted<bool>())
    };

    var expected = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + TimeSpan.FromMinutes(1).Ticks + NotificationOffset + 1, Notification.CreateOnNext(true)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + TimeSpan.FromMinutes(1).Ticks + NotificationOffset + 2, Notification.CreateOnCompleted<bool>())
    };

    var xs = scheduler
        .CreateColdObservable(source)
        .ThrottleOnTrue(TimeSpan.FromMinutes(1), scheduler);

    var observed = scheduler.Start(() => xs, TimeSpan.FromSeconds(3).Ticks + TimeSpan.FromMinutes(1).Ticks + SchedulerOffset);

    CollectionAssert.AreEqual(expected, observed.Messages);
}

Finally, here's a test showing the "T" values being cancelled out by subsequent "F" values:

/// <summary>
/// source:   F---T---F-C
/// expected: -F-------F-C
/// </summary>
[Test]
public void ShouldIgnoreTrueWhenFollowedByFalseWithinDelay()
{
    var scheduler = new TestScheduler();

    var source = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(1).Ticks, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(2).Ticks, Notification.CreateOnNext(true)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + 1, Notification.CreateOnCompleted<bool>())
    };

    var expected = new[]
    {
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(1).Ticks + NotificationOffset + 1, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + NotificationOffset + 1, Notification.CreateOnNext(false)),
        new Recorded<Notification<bool>>(TimeSpan.FromSeconds(3).Ticks + NotificationOffset + 2, Notification.CreateOnCompleted<bool>())
    };

    var xs = scheduler
        .CreateColdObservable(source)
        .ThrottleOnTrue(TimeSpan.FromMinutes(1), scheduler);

    var observed = scheduler.Start(() => xs, TimeSpan.FromSeconds(3).Ticks + TimeSpan.FromMinutes(1).Ticks + SchedulerOffset);

    CollectionAssert.AreEqual(expected, observed.Messages);
}

Pretty sure it's what you were looking for but as your initial marble diagram was a little self-contradictory I can't be 100% sure. Either way, I hope it helps.

Your marble diagram and the text of the description of the problem don't seem to match in my mind. Also the code that you've written doesn't seem to match either of them!

So I've taken the view that you want to always and immediately produce an false values that are emitted. I've also taken the view that when a true comes in you will delay by the delay parameter and only output the true if no other value comes in the meanwhile. If one does you just follow the same rules.

This is based mostly on your code.

You didn't say why you thought your code wasn't working, so I tested it.

Here's my code that turns a marble diagram into an observable and then runes that through your operator:

var marble = "F---T---F---T----T---T---F---C";

Observable
    .Generate(
        0,
        x => marble[x] != 'C',
        x => x + 1,
        x => marble[x] == '-' ? (bool?)null : (marble[x] == 'T' ? true : false),
        x => TimeSpan.FromSeconds(1.0))
    .Where(x => x != null)
    .Select(x => x.Value)
    .ThrottleOnTrue(TimeSpan.FromSeconds(5.0))
    .Timestamp()

Your code produced this:

2020/06/15 01:16:23 +00:00 False 
2020/06/15 01:16:31 +00:00 False 
2020/06/15 01:16:40 +00:00 True 
2020/06/15 01:16:48 +00:00 False 

But the observable never ended.

I'd suggest re-writing your code like this:

public static IObservable<bool> ThrottleOnTrue(this IObservable<bool> source, TimeSpan delay) =>
    source
        .Select(value =>
            Observable
                .Return(value)
                .Delay(value ? delay : TimeSpan.Zero))
        .Switch();

With that I get the same result, but it completes. Hopefully you can work with that.