Наблюдаемое дросселирование

Question

Я пытаюсь реализовать дросселирование событий с реактивными расширениями. У меня есть система, где события могут быть подняты с высокой частотой для конкретного пользователя или другого типа объекта. Мне нужно отложить событие на определенное количество времени, и как только истечет время ожидания, поднимите событие с последним значением.

Что я сделал это

private Subject<int> userBalanceObservable = new Subject<int>(); 
userBalanceObservable.Sample(TimeSpan.FromSeconds(sampleSeconds)) 
      .Subscribe(sample => OnRaiseBalanceEvent(sample)); 

, когда происходит событие

userBalanceObservable.OnNext(userId); 

Редактировать

Проблема с этим подходом является то, что событие вызывается для последнего значения, переданного to OnNext, то, что мне действительно нужно, это иметь dellay для каждого значения, переданного OnNext.

Например OnNext (1), OnNext (2), OnNext (3) я должен был бы иметь отложенный вызов для 1,2,3 вместо я получаю только последнее значение, которое 3.

Answer 1

Sample публикует последнее значение каждый раз, когда удаляется интервал выборки. Если это то, что вам нужно, тогда это нормально. Посмотрите на http://www.introtorx.com/content/v1.0.10621.0/13_TimeShiftedSequences.html#Sample Для получения дополнительной информации + другие способы замедления темпа выбросов.

Что касается новой информации о вашем вопросе:

Если вы хотите, чтобы излучать все значения после определенного тайм-аут был поражен вы можете сгруппировать их, пока ваш тайм-аут не ударил (Примечание: Вы можете запустить из памяти, если вы продолжаете добавлять события без вашего ожидания когда-либо удара из-за frequencey выбросов событий)

вы можете создать буфер, который не заполняет до образования Debounce хитов тайм-аута, увидеть этот ответ на SO для указателей: How to implement buffering with timeout in RX

Answer 2

Не просто buffering работы? Единственная «проблема» в том, что OnRaiseBalanceEvent должен работать со списком, а не только одно значение, но все проблемы в информатике может быть решены другим уровнем косвенности;)

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Threading; 
using System.Reactive.Subjects; 
using System.Reactive.Linq; 

namespace ConsoleApplication1 
{ 
    class Program 
    { 
    static void Main(string[] args) 
    { 

     Subject<int> userBalanceObservable = new Subject<int>(); 
    userBalanceObservable.Buffer(TimeSpan.FromSeconds(2)) //get List of items 
        .Subscribe(sampleList => ProcessSamples(sampleList)); 

     int cont = 0; 

     while (!Console.KeyAvailable) 
     { 
     userBalanceObservable.OnNext(cont); 
     cont++; 
     userBalanceObservable.OnNext(cont); 
     cont++; 
     Thread.Sleep(1000); 
     } 

    } 

    private static void ProcessSamples(IList<int> sampleList) 
    { 
     Console.WriteLine("[{0}]", string.Join(", ", sampleList.ToArray())); 
    } 

    } 
}