باستخدام امتدادات تفاعلية للتمسك بحدث آخر في سلسلة من الأحداث حتى X Seconds بعد استلام الحدث الأخير

Question

أحاول أن أفهم كيف يمكن تنفيذ السيناريو التالي باستخدام الامتدادات التفاعلية. تتضمن الحلول الأخرى التي كنت أبحث عنها إدارة سلسلة من الموقتات ... شيء أتطلع إلى تجنبه إن أمكن.

سيناريو
دفق من "ActionRequests" يحدث. تدفق ActionRequest في مصادر كائن مختلفة. أنا مهتم فقط باختيار أحدث ActionRequest بمجرد أن تتوقف الطلبات من الوصول من مصدر محدد لفترة X Seconds

حتى كسر ذلك.

يتم استلام الأحداث بشكل مستمر من مصادر كائن مختلفة.

يتم الإشارة إلى مصادر الكائن المختلفة من قبل GUID - راجع فئة ActionRequests

بعد X عدد الثواني من الصمت من مصدر الكائن يتم إخراج أحدث حدث.

يمكن التخلص من الرسائل القديمة بمجرد استلام إجراءات الفائدة الحديثة

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 بقلم ريتشارد هاين.

نسخة قصيرة من الإجابة أدناه. تمت إضافة عينة أكثر اكتمالا إلى الخيط في منتديات RX.

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