Wsrtjtyk

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?

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.

@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));

    }

}