Как регулировать поток событий с помощью RX?

Хорошо,

у вас есть 3 сценария:

1) Я хотел бы получать одно значение потока событий каждую секунду. Это означает: если он производит больше событий в секунду, вы получите всегда больший буфер.

observableStream.Throttle(timeSpan)

2) Я хотел бы получить последнее событие, которое было произведено до того, как произойдет второе. означает: другие события отбрасываются.

observableStream.Sample(TimeSpan.FromSeconds(1))

3) вы хотите получить все события, которые произошли в последнюю секунду. и так каждую секунду

observableStream.BufferWithTime(timeSpan)

4) вы хотите выбрать то, что происходит между секундами со всеми значениями, пока не пройдет секунда, и ваш результат будет возвращен

observableStream.CombineLatest(Observable.Interval(1000), selectorOnEachEvent)

Это то, что я опубликовал в качестве ответа на этот вопрос на форуме Rx :

ОБНОВЛЕНИЕ : Вот новая версия, которая больше не задерживает пересылку событий, когда события происходят с разницей во времени более одной секунды:

public static IObservable<T> ThrottleResponsive3<T>(this IObservable<T> source, TimeSpan minInterval)
{
    return Observable.CreateWithDisposable<T>(o =>
    {
        object gate = new Object();
        Notification<T> last = null, lastNonTerminal = null;
        DateTime referenceTime = DateTime.UtcNow - minInterval;
        var delayedReplay = new MutableDisposable();
        return new CompositeDisposable(source.Materialize().Subscribe(x =>
        {
            lock (gate)
            {
                var elapsed = DateTime.UtcNow - referenceTime;
                if (elapsed >= minInterval && delayedReplay.Disposable == null)
                {
                    referenceTime = DateTime.UtcNow;
                    x.Accept(o);
                }
                else
                {
                    if (x.Kind == NotificationKind.OnNext)
                        lastNonTerminal = x;
                    last = x;
                    if (delayedReplay.Disposable == null)
                    {
                        delayedReplay.Disposable = Scheduler.ThreadPool.Schedule(() =>
                        {
                            lock (gate)
                            {
                                referenceTime = DateTime.UtcNow;
                                if (lastNonTerminal != null && lastNonTerminal != last)
                                    lastNonTerminal.Accept(o);
                                last.Accept(o);
                                last = lastNonTerminal = null;
                                delayedReplay.Disposable = null;
                            }
                        }, minInterval - elapsed);
                    }
                }
            }
        }), delayedReplay);
    });
}

Это была моя предыдущая попытка:

var source = Observable.GenerateWithTime(1, 
    x => x <= 100, x => x, x => TimeSpan.FromMilliseconds(1), x => x + 1)
    .Timestamp();

source.Publish(o =>
    o.Take(1).Merge(o.Skip(1).Sample(TimeSpan.FromSeconds(1)))
).Run(x => Console.WriteLine(x));

Пробовали ли вы метод расширения Throttle?

Из документации:

Игнорирует значения из наблюдаемой последовательности, за которыми следует другое значение до dueTime

Мне не совсем ясно, будет ли это делать то, что вы хотите, или нет - в том смысле, что вы хотите игнорировать следующие значения, а не первое... но я бы ожидал, что это будет то, что вы хотите. Попробуйте :)

EDIT: Хммм... нет, я не думаю, что Throttle это то, что нужно, в конце концов. Мне кажется, я понимаю, что вы хотите сделать, но я не вижу ничего в фреймворке для этого. Хотя, вполне возможно, я что-то упустил. Вы спрашивали на форуме Rx? Вполне возможно, что если этого нет сейчас, то они с радостью добавят это :)

Я подозреваю, что вы можете сделать это хитро с SkipUntil и SelectMany каким-то образом... но я думаю, что это должно быть в собственном методе.

Хорошо, вот одно решение. Мне оно не очень нравится, но... ладно.

Спасибо Джону за то, что указал мне на SkipWhile, и cRichter за BufferWithTime. Спасибо, ребята.

static void Main(string[] args)
{
    Console.WriteLine("Running...");

    var generator = Observable
        .GenerateWithTime(1, x => x <= 100, x => x, x => TimeSpan.FromMilliseconds(1), x => x + 1)
        .Timestamp();

    var bufferedAtOneSec = generator.BufferWithTime(TimeSpan.FromSeconds(1));

    var action = new Action<Timestamped<int>>(
        feed => Console.WriteLine("Observed {0:000}, generated at {1}, observed at {2}",
                                  feed.Value,
                                  feed.Timestamp.ToString("mm:ss.fff"),
                                  DateTime.Now.ToString("mm:ss.fff")));

    var reactImmediately = true;
    bufferedAtOneSec.Subscribe(list =>
                                   {
                                       if (list.Count == 0)
                                       {
                                           reactImmediately = true;
                                       }
                                       else
                                       {
                                           action(list.Last());
                                       }
                                   });
    generator
        .SkipWhile(item => reactImmediately == false)
        .Subscribe(feed =>
                       {
                           if(reactImmediately)
                           {
                               reactImmediately = false;
                               action(feed);
                           }
                       });

    Console.ReadKey();
}

То, что вы ищете, это CombineLatest.

public static IObservable<TResult> CombineLatest<TLeft, TRight, TResult>(
    IObservable<TLeft> leftSource,
    IObservable<TRight> rightSource,
    Func<TLeft, TRight, TResult> selector
)

который объединяет 2 наблюдаемые таблицы и возвращает все значения, когда селектор (время) имеет значение.

edit: john is right, that is perhaps not the preferred solution