A way to push buffered events in even intervals

by

It’s actually tricker than it sounds.

Using Delay doesn’t work because the values will still happen in bulk, only slightly delayed.

Using Interval with either CombineLatest or Zip doesn’t work, since the former will cause source values to be skipped and the latter will buffer interval values.

I think the new Drain operator (added in 1.0.2787.0), combined with Delay should do the trick:

source.Drain(x => Observable.Empty<int>().Delay(TimeSpan.FromSeconds(1)).StartWith(x));

The Drain operator works like SelectMany, but waits until the previous output completes before calling the selector with the next value. It’s still not exactly what you are after (the first value in a block will also be delayed), but it’s close: The usage above matches your marble diagram now.

Edit: Apparently the Drain in the framework doesn’t work like SelectMany. I’ll ask for some advice in the official forums. In the meantime, here’s an implementation of Drain that does what you’re after:

Edit 09/11: Fixed errors in implementation and updated usage to match your requested marble diagram.

public static class ObservableDrainExtensions
{
    public static IObservable<TOut> Drain<TSource, TOut>(this IObservable<TSource> source, 
        Func<TSource, IObservable<TOut>> selector)
    {
        return Observable.Defer(() =>
        {
            BehaviorSubject<Unit> queue = new BehaviorSubject<Unit>(new Unit());

            return source
                .Zip(queue, (v, q) => v)
                .SelectMany(v => selector(v)
                    .Do(_ => { }, () => queue.OnNext(new Unit()))
                );
        });
    }
}

More Related Contents:

  1. ObserveOn and SubscribeOn – where the work is being done
  2. How can I implement an exhaustMap handler in Rx.Net?
  3. Good introduction to the .NET Reactive Framework [closed]
  4. Does reactive extensions support rolling buffers?
  5. Reactive Extensions for .NET (Rx): Take action once all events are completed
  6. Best way to reverse a string
  7. Difference in months between two dates
  8. Best way to remove duplicate entries from a data table
  9. Is there a built-in method to compare collections?
  10. Task continuation on UI thread
  11. How to compare values of generic types?
  12. String interning in .Net Framework – What are the benefits and when to use interning
  13. Generate N random and unique numbers within a range
  14. Win32Exception Not enough storage is available to process this command
  15. Convert string to Color in C#
  16. Use linq to generate direct update without select
  17. When is CanExecute called?
  18. How to use the .NET Timer class to trigger an event at a specific time?
  19. Using extension methods in .NET 2.0?
  20. Exception.Message vs Exception.ToString()
  21. Can I use ASP.NET MVC together with regular ASP.NET Web forms
  22. Creating local user account c# and .NET 2.0
  23. Simplest way to make cross-appdomain call?
  24. Obtaining only the filename when using OpenFileDialog property “FileName”
  25. Finding information about all serial devices connected through USB in C#
  26. Adding a select all shortcut (Ctrl + A) to a .net listview?
  27. Heap versus Stack allocation implications (.NET)
  28. How to refresh datagridview when closing child form?
  29. Send message to a Windows process (not its main window)
  30. Passing a variable between Windows Forms forms

Leave a Comment