Rx: Наблюдаемые "повторяемы", как IEnumerable, и если нет, как работает этот код?

Question

Вчера я смотрел скринкаст Написание вашего первого приложения Rx (на канале 9), где Уэс Дайер показывает, как реализовать Drag 'n' Drop с помощью Reactive Extensions (Rx ) . Что-то, чего я до сих пор не понимаю:

В конце экрана Уэс Дайер пишет следующее:

var q = from start in mouseDown
        from delta in mouseMove.StartsWith(start).Until(mouseUp)
                       .Let(mm => mm.Zip(mm.Skip(1), (prev, cur) =>
                           new { X = cur.X - prev.X, Y = cur.Y - prev.Y }))
        select delta;

Вкратце, q - это наблюдаемая, которая передает дельты координат перемещения мыши своим подписчикам.

Что я не понимаю, так это то, как mm.Zip(mm.Skip(1), ...) может работать!?

Насколько я знаю, IObservable не является перечислимым в том смысле, как IEnumerable. Благодаря природе «1022» «тянуть» его можно повторять снова и снова, всегда получая одни и те же элементы. (По крайней мере это должно иметь место для всех хорошо перечисляемых перечислимых значений.) IObservable работает по-другому. Элементы передаются подписчикам один раз, и на этом все. В приведенном выше примере движения мыши представляют собой единичные инциденты, которые не могут повторяться без записи в память.

Итак, как может сработать комбинация .Zip с .Skip(1), поскольку события мыши, над которыми они работают, являются единичными, неповторяющимися инцидентами? Разве эта операция не требует, чтобы mm "просматривался" независимо дважды?

---

Для справки вот подпись метода Observable.Zip:

public static IObservable<TResult> Zip <TLeft, TRight, TResult>
(
    this IObservable<TLeft>       leftSource,     //  = mm
    IObservable<TRight>           rightSource,    //  = mm.Skip(1)
    Func<TLeft, TRight, TResult>  selector
)

---

P.S .: Я только что увидел, что есть еще один скринкаст с оператором Zip , который довольно проницательный.

Answer 1

Разве эта операция не требует, чтобы мм "просматривался" независимо дважды?

Это фактически ответ на ваш вопрос: вы можете подписаться на одну и ту же последовательность IObservable несколько раз.

mm.Skip(1) подписывается на mm и скрывает первое значение для своих подписчиков. Zip является подписчиком mm.Skip(1) и mm. Поскольку mm дал на одно большее значение, чем mm.Skip(1), Zip внутренне буферизует последнее событие mousemove из mm все время, чтобы связать его со следующим будущим событием mousemove. Функция выбора может затем выбирать дельту между обоими.

Еще одна вещь, которую вы должны заметить (это реальный ответ на заголовок вашего вопроса), что это Observable.FromEvent - IObservable является горячим наблюдаемым и, следовательно, не повторяется . Но существуют холодные наблюдаемые, которые на самом деле повторяемые , как Observable.Range (0,10). В последнем случае каждый подписчик получит одинаковые 10 событий, поскольку они генерируются независимо для каждого подписчика. Для событий mousemove это не так (вы не получите события перемещения мыши из прошлого). Но поскольку Zip одновременно подписывается на правую и левую последовательности, в этом случае, вероятно, то же самое.

P.S .: Вы также можете создать горячее / не повторяемое IEnumerable: Нет необходимости возвращать одинаковые значения для каждого перечислителя. Например, вы можете создать IEnumerable, который ждет, пока не произойдет событие mousemove, а затем выдаст событие. В этом случае перечислитель всегда будет блокировать (плохой дизайн), но это будет возможно. ;)

Answer 2

Aha!Zip скринкаст , о котором я упоминал в PS, дал мне жизненно важную подсказку: Zip "запоминает" предметы, чтобы учесть тот факт, что предметы могут прибыть из одного наблюдаемого раньше, чем из другого.Я попытаюсь ответить на мой вопрос, я надеюсь, что кто-то может исправить меня, если я ошибаюсь.

Zip объединяет входные данные из двух наблюдаемых последовательностей, подобных этой (буквы и цифры являются «событиями»):

mm                        ----A---------B-------C------D-----------E----->
                              |         |       |      |           |
                              |         |       |      |           |
mm.Skip(1)                ----+---------1-------2------3-----------4----->
                              |         |       |      |           |
                              |         |       |      |           |
mm.Zip(mm.Skip(1), ...)   ----+--------A,1-----B,2----C,3---------D,4---->

И это действительно должно делать внутреннюю буферизацию.В коде, который я разместил, mm является реальной, «живой» наблюдаемой.mm.Skip(1) - это нечто вроде конечного автомата, полученного из него.Ответ Алекса Павена кратко объясняет, как это работает.

Итак, mm.Zip(mm.Skip(1), ...) действительно смотрит на mm дважды, один раз напрямую и один раз через фильтр Skip(n).И поскольку наблюдаемые не являются повторяемыми последовательностями, он выполняет внутреннюю буферизацию, чтобы учесть тот факт, что одна последовательность даст элементы быстрее, чем другая.

(Я быстро взглянул на источник Rx с помощью .NET Reflector и действительно,Zip включает Queue.)

Answer 3

Элементы передаются подписчикам один раз, и это было все.

Да, один элемент передается один раз, но элемент является одним из «последовательности» событий.Последовательность все еще является последовательностью.Вот почему Skip работает - он пропускает один элемент, а затем, когда приходит следующий, обрабатывает его (не пропускает).