Подписчики пропускают сообщения; Является ли это ошибкой с Rx или ее ошибкой?

Question

У меня есть одно приложение окно WPF со следующим конструктором

 numbers = Observable.Generate(DateTime.Now, 
             time => true, 
             time => DateTime.Now, 
             time => { return new  Random(DateTime.Now.Millisecond).NextDouble() * 99 + 2; }, 
             time => TimeSpan.FromSeconds(1.0)); 


     numbers.ObserveOnDispatcher() 
      .Subscribe(s => list1.Items.Add(s.ToString("##.00"))); 

     numbers.Where(n => n < 10).ObserveOnDispatcher(). 
      Subscribe(s => list2.Items.Add(s.ToString("##.00"))); 

Теперь вот скриншот списков - Обратите внимание на 3,76 отсутствует в списке слева ... Это поведение прерывистым.

Answer 1

Короткий ответ в том, что вы делаете это неправильно. Rx работает отлично.

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

Таким образом, в вашем случае у вас есть два экземпляра этой операционной последовательности:

var numbers = 
    Observable 
     .Generate(
      DateTime.Now, 
      time => true, 
      time => DateTime.Now, 
      time => new Random(DateTime.Now.Millisecond) 
       .NextDouble() * 99 + 2, 
      time => TimeSpan.FromSeconds(1.0)); 

Теперь, поскольку вы подписавшись на этой наблюдаемой дважды в непосредственной последовательности два экземпляра этой наблюдаемой будет пытаться генерировать значения почти в то же время. Значит, значение DateTime.Now.Millisecond было бы одинаковым в большинстве случаев, но теперь всегда. Значение, возвращаемое с new Random(x).NextDouble(), одинаково для одного и того же x, поэтому поэтому большую часть времени вы получаете одинаковое значение из двух экземпляров наблюдаемого. Это просто, когда DateTime.Now.Millisecond отличается тем, что вы получаете два разных значения, и кажется, что у подписчиков отсутствуют значения.

---

Вот альтернативная версия, которая должна работать, как вы ожидали изначально: «Является ли это ошибка с Rx или его мне делать это неправильно»

var rnd = new Random((int)DateTime.Now.Ticks); 

var numbers = 
    Observable 
     .Generate(0, n => true, n => 0, 
      n => rnd.NextDouble() * 99 + 2, 
      n => TimeSpan.FromSeconds(1.0)); 

var publishedNumbers = numbers.Publish(); 

publishedNumbers 
    .ObserveOnDispatcher() 
    .Subscribe(s => list1.Items.Add(s.ToString("##.00"))); 

publishedNumbers 
    .Where(n => n < 10) 
    .ObserveOnDispatcher() 
    .Subscribe(s => list2.Items.Add(s.ToString("##.00"))); 

publishedNumbers.Connect();