With Rx, how do I ignore all-except-the-latest value when my Subscribe method is running

by

Here is a method that is similar to Dave’s but uses Sample instead (which is more appropriate than buffer). I’ve included a similar extension method to the one I added to Dave’s answer.

The extension:

public static IDisposable SubscribeWithoutOverlap<T>(this IObservable<T> source, Action<T> action)
{
    var sampler = new Subject<Unit>();

    var sub = source.
        Sample(sampler).
        ObserveOn(Scheduler.ThreadPool).
        Subscribe(l =>
        {
            action(l);
            sampler.OnNext(Unit.Default);
        });

    // start sampling when we have a first value
    source.Take(1).Subscribe(_ => sampler.OnNext(Unit.Default));

    return sub;
}

Note that it’s simpler, and there is no ’empty’ buffer that’s fired. The first element that is sent to the action actually comes from the stream itself.

Usage is straightforward:

messages.SubscribeWithoutOverlap(n =>
{
    Console.WriteLine("start: " + n);
    Thread.Sleep(500);
    Console.WriteLine("end: " + n);
});

messages.Subscribe(x => Console.WriteLine("source: " + x)); // for testing

And results:

source: 0
start: 0
source: 1
source: 2
source: 3
source: 4
source: 5
end: 0
start: 5
source: 6
source: 7
source: 8
source: 9
source: 10
end: 5
start: 10
source: 11
source: 12
source: 13
source: 14
source: 15
end: 10

More Related Contents:

  1. ObserveOn and SubscribeOn – where the work is being done
  2. How to throttle event stream using RX?
  3. How can I implement an exhaustMap handler in Rx.Net?
  4. Why are Subjects not recommended in .NET Reactive Extensions?
  5. How can I see what my reactive extensions query is doing?
  6. Write an Rx “RetryAfter” extension method
  7. Good introduction to the .NET Reactive Framework [closed]
  8. Does reactive extensions support rolling buffers?
  9. Rx: How can I respond immediately, and throttle subsequent requests
  10. A way to push buffered events in even intervals
  11. What are the default Schedulers for each observable operator?
  12. How can I alternately buffer and flow a live data stream in Rx
  13. What’s a good way to run periodic tasks using Rx, with a single concurrent execution restriction?
  14. How to call back async function from Rx Subscribe?
  15. Reactive Extensions for .NET (Rx): Take action once all events are completed
  16. Pause and Resume Subscription on cold IObservable
  17. Interfaces — What’s the point?
  18. How to copy data to clipboard in C#
  19. C# WPF Navigation Between Pages (Views)
  20. Does page reload ever cause post?
  21. Right way to dispose Image/Bitmap and PictureBox
  22. How do I get the currently loggedin Windows account from an ASP.NET page?
  23. How to get the most common value in an Int array? (C#)
  24. C# Empty Statement
  25. How can I determine whether Console.Out has been redirected to a file?
  26. How to give my C# app administrative rights? manifest file
  27. LINQ query — Data aggregation (group adjacent)
  28. Compare two Color objects
  29. C# Dynamic Keyword — Run-time penalty?
  30. Json.Net deserialize out of memory issue

Leave a Comment