RX - How to subscribe for condition state, but only when this state doesn't change for x period of time?

Question

Do you know how to subscribe for condition state that stay for x time?

For example if I have BehaviorSubject<int> that represent int value between 0 - 100, and this value is changing over time, I want to subscribe when this value is under 50 for 10 seconds continuously.

if the value change back to above 50 for a moment and then down 50 again, I want to count again for 10 seconds. How can I do this?

Many thanks!

Answer 1

Here's a fairly neat observable query that seemed to work properly for me:

var below50longerthan10seconds =
    subject
        .Select(x => x < 50)
        .DistinctUntilChanged()
        .Select(x =>
            Observable.Delay(
                Observable.Return(x),
                TimeSpan.FromSeconds(10.0)))
        .Switch()
        .Where(x => x)
        .Select(x => Unit.Default);

Here's the breakdown.

Change the values from 0 to 100 to true when less than 50 and false otherwise:

        .Select(x => x < 50)

Only keep the actual changes between true & false:

        .DistinctUntilChanged()

Project the value into a new observable that is delayed for 10 seconds:

        .Select(x =>
            Observable.Delay(
                Observable.Return(x),
                TimeSpan.FromSeconds(10.0)))

Now perform a .Switch() - if a new x comes before the delayed observable then it is ignored because a new delayed observable is coming:

        .Switch()

Select only the true values meaning when the original stream was below 50:

        .Where(x => x)

Select Unit.Default simply because it is weird to have a stream of true values without any false values:

        .Select(x => Unit.Default);

So now you have an IObservable<Unit> that releases a new value when the original stream produces a value less then 50 and does not produce a value greater than or equal to 50 within 10 seconds.

Is that what you wanted?

Answer 2

I hope I don't miss something but I think the easiest way is this:

BehaviorSubject<int> value = new BehaviorSubject<int>(0);

value.Select(v => v < 50).DistinctUntilChanged().Throttle(TimeSpan.FromSeconds(10))
.Where(x => x).Subscribe(b => DoSomething());

the answer of Enigmativity gave me the start how to get handle this . Thanks!