Использование реактивных расширений для удержания самого последнего события в серии событий в течение x секунд после получения самого последнего события

Question

Я пытаюсь понять, как следующий сценарий может быть реализован с использованием реактивных расширений.Другие решения, на которые я смотрел, включают управление серией таймеров ... то, чего я стараюсь избегать, если это возможно.

Сценарий
Поток "ActionRequests" происходит,Поток ActionRequest поступает из разных источников объектов.Я заинтересован только в том, чтобы отбирать самые последние запросы ActionRequest, как только запросы перестали поступать из определенного источника в течение периода x секунд

.

События постоянно принимаются из разных источников объектов. Разные источники объектов обозначаются guid - см. Класс ActionRequests Через x секунд тишины от источника объекта последнее событие выталкивается. Старые сообщения могут быть отброшены, как только будет получен более свежий интересный запрос ActionRequest

public class ActionRequest
{
    public ActionRequest(Guid quoteId, DateTime now)
    {
        QuoteId = quoteId;
        RequestTimestamp = now;
    }

    public Guid QuoteId { get; set; }
    public DateTime RequestTimestamp { get; set; }

}

Любые мысли по этому поводу приветствуются.

Answer 1

Ответ был предоставлен в RX Forums Ричардом Хейном.

Краткая версия ответа ниже. Более полный образец был добавлен в ветку на форумах RX.

var throttled = observable.GroupBy(tar => tar.Value.QuoteId).SelectMany(g => g.Throttle(TimeSpan.FromMilliseconds(1500)));