What’s a good way to run periodic tasks using Rx, with a single concurrent execution restriction?

by

You are on the right track, you can use Select + Concat to flatten out the observable and limit the number of inflight requests (Note: if your task takes longer than the interval time, then they will start to stack up since they can’t execute fast enough):

var source = Observable.Interval(TimeSpan.FromMilliseconds(100))
          //I assume you are doing async work since you want to limit concurrency
          .Select(_ => Observable.FromAsync(() => DoSomethingAsync()))
          //This is equivalent to calling Merge(1)
          .Concat();

source.Subscribe(/*Handle the result of each operation*/);

More Related Contents:

  1. How can I implement an exhaustMap handler in Rx.Net?
  2. How to call back async function from Rx Subscribe?
  3. ObserveOn and SubscribeOn – where the work is being done
  4. How to throttle event stream using RX?
  5. Why are Subjects not recommended in .NET Reactive Extensions?
  6. How can I see what my reactive extensions query is doing?
  7. Write an Rx “RetryAfter” extension method
  8. Good introduction to the .NET Reactive Framework [closed]
  9. With Rx, how do I ignore all-except-the-latest value when my Subscribe method is running
  10. Does reactive extensions support rolling buffers?
  11. Rx: How can I respond immediately, and throttle subsequent requests
  12. A way to push buffered events in even intervals
  13. What are the default Schedulers for each observable operator?
  14. How can I alternately buffer and flow a live data stream in Rx
  15. Reactive Extensions for .NET (Rx): Take action once all events are completed
  16. Pause and Resume Subscription on cold IObservable
  17. Create dynamic variable name
  18. Json.Net Serialization of Type with Polymorphic Child Object
  19. #if DEBUG vs. Conditional(“DEBUG”)
  20. How to detect installed version of MS-Office?
  21. ToList()– does it create a new list?
  22. How to stop BackgroundWorker correctly
  23. How to get timestamp of tick precision in .NET / C#?
  24. How to avoid System.IO.PathTooLongException?
  25. High performance TCP server in C#
  26. Adding scripting functionality to .NET applications
  27. Cannot use a lambda expression as an argument to a dynamically dispatched operation without first casting it to a delegate or expression tree type
  28. Validate Enum Values
  29. RowSpan does not work in iTextSharp?
  30. Examples of IoC Containers [closed]

Leave a Comment