CODEWITHSUNDEEP
reactive-programmingsystem.reactiverx.net
Evgeny
Evgeny

Reputation: 641

Rx.NET Take an element and subscribe again after some time

I need a kind of throttling, which works a bit differently. I need to get an element from a sequence, unsubscribe and subscribe again in 1 sec. In other words, I want to ignore all elements during 1 sec after a first element is taken:

Input:  (1) -100ms- (2) -200ms- (3) -1_500ms- (4) -1_000ms- (5) -500ms- (6) ...
Output: (1) --------------------------------- (4) --------- (5) ----------- ...

How can achieve this simple thing with Rx.NET?

Upvotes: 0

Views: 151

Answers (2)

Shlomo
Shlomo

Reputation: 14370

@Enigmativity's answer doesn't fit the specs exactly. It may work for what you want though.

His answer defines 1-second windows, and takes the first from each of those windows. This doesn't guarantee you one second of silence between items though. Consider this case:

t     : ---------1---------2---------3
source: ------1---2------3---4----5--|
window: ---------|---------|---------|
spec  : ------1----------3-----------|
enigma: ------1---2----------4-------|

The answer implies you want one second of nothing after item 1. The next item after that is 3, with silence then through the end. Here's that test coded up:

var scheduler = new TestScheduler();
var source = scheduler.CreateColdObservable<int>(
    RxTest.OnNext(700.MsTicks(),  1),
    RxTest.OnNext(1100.MsTicks(), 2),
    RxTest.OnNext(1800.MsTicks(), 3),
    RxTest.OnNext(2200.MsTicks(), 4),
    RxTest.OnNext(2600.MsTicks(), 5),
    RxTest.OnCompleted<int>(3000.MsTicks())
);

var expectedResults = scheduler.CreateHotObservable<int>(
    RxTest.OnNext(700.MsTicks(),  1),
    RxTest.OnNext(1800.MsTicks(), 3),
    RxTest.OnCompleted<int>(3000.MsTicks())
);

var target = source
    .Window(() => Observable.Timer(TimeSpan.FromSeconds(1.0), scheduler))
    .SelectMany(xs => xs.Take(1));

var observer = scheduler.CreateObserver<int>();
target.Subscribe(observer);
scheduler.Start();
ReactiveAssert.AreElementsEqual(expectedResults.Messages, observer.Messages);

I think the best way around this is a Scan-based solution with timestamps. You basically hold the last legit message in memory with a timestamp, and if the new message is one second older, emit. Otherwise, don't:

public static IObservable<T> TrueThrottle<T>(this IObservable<T> source, TimeSpan span)
{
    return TrueThrottle<T>(source, span, Scheduler.Default);
}

public static IObservable<T> TrueThrottle<T>(this IObservable<T> source, TimeSpan span, IScheduler scheduler)
{
    return source
        .Timestamp(scheduler)
        .Scan(default(Timestamped<T>), (state, item) => state == default(Timestamped<T>) || item.Timestamp - state.Timestamp > span
            ? item
            : state
        )
        .DistinctUntilChanged()
        .Select(t => t.Value);
}

Note: Test code uses Nuget Microsoft.Reactive.Testing and the following helper class:

public static class RxTest
{
    public static long MsTicks(this int i)
    {
        return TimeSpan.FromMilliseconds(i).Ticks;
    }

    public static Recorded<Notification<T>> OnNext<T>(long msTicks, T t)
    {
        return new Recorded<Notification<T>>(msTicks, Notification.CreateOnNext(t));
    }

    public static Recorded<Notification<T>> OnCompleted<T>(long msTicks)
    {
        return new Recorded<Notification<T>>(msTicks, Notification.CreateOnCompleted<T>());
    }

    public static Recorded<Notification<T>> OnError<T>(long msTicks, Exception e)
    {
        return new Recorded<Notification<T>>(msTicks, Notification.CreateOnError<T>(e));
    }
}

Upvotes: 1

Enigmativity
Enigmativity

Reputation: 117175

Try this:

Input
    .Window(() => Observable.Timer(TimeSpan.FromSeconds(1.0)))
    .SelectMany(xs => xs.Take(1));

Here's a test:

var query =
    Observable
        .Interval(TimeSpan.FromSeconds(0.2))
        .Window(() => Observable.Timer(TimeSpan.FromSeconds(1.0)))
        .SelectMany(xs => xs.Take(1));

This produced:

0 
5 
10 
14 
19 
24 
29 
34 
39

The jump from 10 to 14 is just a result of using multiple threads and not an error in the query.

Upvotes: 2

Related Questions