Implement a real throttle (not debounce, like the existing throttle)

[cmeeren]

Currently there is no operator to ignore elements from an observable sequence that follow a non-ignored element within a specified relative time duration. This is what would normally be called Throttle, while the existing Throtte operator is really a debounce operator (see e.g. The Difference Between Throttling and Debouncing).

Here is a sample implementation of the "real" throttle operator both with and without scheduler:

public static IObservable<TSource> AtMostOnceEvery<TSource>(
    this IObservable<TSource> source,
    TimeSpan dueTime)
{
    IObservable<TSource> delay = Observable.Empty<TSource>().Delay(dueTime);
    return source.Take(1).Concat(delay).Repeat();
}

public static IObservable<TSource> AtMostOnceEvery<TSource>(
    this IObservable<TSource> source,
    TimeSpan dueTime,
    IScheduler scheduler)
{
    IObservable<TSource> delay = Observable.Empty<TSource>().Delay(dueTime, scheduler);
    return source.Take(1).Concat(delay).Repeat();
}

[akarnokd]

That won't work if the source is cold because repeat will resubscribe and you get the very first element all over.

This should work for both cold and hot sources :

    class Program
    {
        static void Main(string[] args)
        {
            new int[] { 100, 200, 300, 400, 500, 600, 1000 }.ToObservable()
                .SelectMany(v => Observable.Timer(TimeSpan.FromMilliseconds(v)).Select(w => v))
                .ThrottleFirst(TimeSpan.FromMilliseconds(210), Scheduler.Default)
                .Subscribe(Console.WriteLine, () => Console.WriteLine("Done"));

            Console.ReadLine();
        }
    }

    static class OX
    {
        internal static IObservable<T> ThrottleFirst<T>(this IObservable<T> source, TimeSpan delay, IScheduler scheduler)
        {
            return source.Publish(o =>
            {
                return o.Take(1)
                .Concat(o.IgnoreElements().TakeUntil(Observable.Return(default(T)).Delay(delay, scheduler)))
                .Repeat()
                .TakeUntil(o.IgnoreElements().Concat(Observable.Return(default(T))));
            });
        }
    }

[cmeeren]

Thanks. I don't understand what's happening there because I haven't used Rx all that much, but I'd just like to add that if this is implemented, perhaps the word "debounce" should appear in the Throttle documentation.

[akarnokd]

See the ReactiveX page of Sample which lists other throttle-like implementations in other languages with marble diagrams: http://reactivex.io/documentation/operators/sample.html

[cmeeren]

Not sure why you refer me to that page. Sample seems different than what I describe: With Sample, an item is picked each X seconds (if a new item exists). If you sample every 10 seconds and an item is pushed 1 second after the previous sample, you'd have to wait 9 seconds for Sample to pick it up. I am talking about immediately acting on an item and then skipping all other items until the delay has completed.

[akarnokd]

You need to browse a little bit on that page:

[cmeeren]

That's exactly what I suggested, isn't it? Possible clarification: When I said

I don't understand what's happening there

I was thinking about the specific implementation you posted above, but no matter - I assume we're on the same page as regards the desired behavior. :)

[Matungos]

I know is offtopic, but could you please share an implementation of throttleWithTimeout??
Because I'm trying to implement the debounce behavior without success.

[sglienke]

@Matungos That is what Throttle already does (it's a bit confusing I have to say that RX.NET calls Debounce Throttle and Throttle Sample)

[Bartolomeus-649]

What's the status of this?

And does all agree on the definitions:
Throttle: Maximum number of calls during a specified time period.
Debounce: Minimum time between calls.

[DavidTangWei]

@akarnokd hello~
Your implementation not work for me. My need is to find or create a rx operator like throttleFirst in java.Rx.

[akarnokd]

@DavidTangWei Here is a native implementation:

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Hello World!");

            new int[] { 100, 200, 300, 400, 500, 600, 1000 }.ToObservable()
                .SelectMany(v => Observable.Timer(TimeSpan.FromMilliseconds(v)).Select(w => v))
                .ThrottleFirst(TimeSpan.FromMilliseconds(210), Scheduler.Default)
                .Subscribe(Console.WriteLine, () => Console.WriteLine("Done"));

            Console.ReadLine();
        }
    }

    static class ObservablesEx
    {
        public static IObservable<T> ThrottleFirst<T>(this IObservable<T> source, 
                TimeSpan timespan, IScheduler timeSource)
        {
            return new ThrottleFirstObservable<T>(source, timeSource, timespan);
        }
    }

    sealed class ThrottleFirstObservable<T> : IObservable<T>
    {
        readonly IObservable<T> source;

        readonly IScheduler timeSource;

        readonly TimeSpan timespan;

        internal ThrottleFirstObservable(IObservable<T> source, 
                  IScheduler timeSource, TimeSpan timespan)
        {
            this.source = source;
            this.timeSource = timeSource;
            this.timespan = timespan;
        }

        public IDisposable Subscribe(IObserver<T> observer)
        {
            var parent = new ThrottleFirstObserver(observer, timeSource, timespan);
            var d = source.Subscribe(parent);
            parent.OnSubscribe(d);
            return d;
        }

        sealed class ThrottleFirstObserver : IDisposable, IObserver<T>
        {
            readonly IObserver<T> downstream;

            readonly IScheduler timeSource;

            readonly TimeSpan timespan;

            IDisposable upstream;

            bool once;

            double due;

            internal ThrottleFirstObserver(IObserver<T> downstream, 
                    IScheduler timeSource, TimeSpan timespan)
            {
                this.downstream = downstream;
                this.timeSource = timeSource;
                this.timespan = timespan;
            }

            public void OnSubscribe(IDisposable d)
            {
                if (Interlocked.CompareExchange(ref upstream, d, null) != null)
                {
                    d.Dispose();
                }
            }

            public void Dispose()
            {
                var d = Interlocked.Exchange(ref upstream, this);
                if (d != null && d != this)
                {
                    d.Dispose();
                }
            }

            public void OnCompleted()
            {
                downstream.OnCompleted();
            }

            public void OnError(Exception error)
            {
                downstream.OnError(error);
            }

            public void OnNext(T value)
            {
                var now = timeSource.Now.ToUnixTimeMilliseconds();
                if (!once)
                {
                    once = true;
                    due = now + timespan.TotalMilliseconds;
                    downstream.OnNext(value);
                } else if (now >= due)
                {
                    due = now + timespan.TotalMilliseconds;
                    downstream.OnNext(value);
                }

            }
        }
    }

[DavidTangWei]

@akarnokd Thanks~ It works. I need to use Rx.net Rx-Main -Version 2.2.5 in order to support .net4.0. So my only approach is to copy this implement to my source code.